Issue:
Bug: [FEATURE] : Need a framework to test the Req, Res and router data which is made through UCS



### Feature Description

There can be possibilities where the req, res or router_data may be differ in case of UCS, so to thoroughly check if these things are correct or not we need a framework.

### Possible Implementation

Implement a Shadow run framework for ucs so that we can test it without bringing it on critical path

### Have you spent some time checking if this feature request has been raised before?

- [x] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [x] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

None

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
Added a fork using tokio::spawn and sent the cloned router data to UCS, 
Added some headers for identification of connector, flow, req-id 
called the validation service to send both router data for comparison.


### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [x] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

Patch:
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index a6b9acd7e3c..6739cde8f8d 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -2274,6 +2274,33 @@ pub enum GatewaySystem {
     #[default]
     Direct,
     UnifiedConnectorService,
+    ShadowUnifiedConnectorService,
+}
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Default,
+    Eq,
+    PartialOrd,
+    Ord,
+    Hash,
+    PartialEq,
+    serde::Deserialize,
+    serde::Serialize,
+    strum::Display,
+    strum::VariantNames,
+    strum::EnumIter,
+    strum::EnumString,
+    ToSchema,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum ExecutionMode {
+    #[default]
+    Primary,
+    Shadow,
 }
 
 /// The type of the payment that differentiates between normal and various types of mandate payments. Use 'setup_mandate' in case of zero auth flow.
diff --git a/crates/common_utils/src/consts.rs b/crates/common_utils/src/consts.rs
index bbb65e1afc6..e2681057f9f 100644
--- a/crates/common_utils/src/consts.rs
+++ b/crates/common_utils/src/consts.rs
@@ -157,6 +157,15 @@ pub const APPLEPAY_VALIDATION_URL: &str =
 /// Request ID
 pub const X_REQUEST_ID: &str = "x-request-id";
 
+/// Flow name
+pub const X_FLOW_NAME: &str = "x-flow";
+
+/// Connector name
+pub const X_CONNECTOR_NAME: &str = "x-connector";
+
+/// Unified Connector Service Mode
+pub const X_UNIFIED_CONNECTOR_SERVICE_MODE: &str = "x-shadow-mode";
+
 /// Chat Session ID
 pub const X_CHAT_SESSION_ID: &str = "x-chat-session-id";
 
diff --git a/crates/common_utils/src/errors.rs b/crates/common_utils/src/errors.rs
index 82028076969..66ae55391df 100644
--- a/crates/common_utils/src/errors.rs
+++ b/crates/common_utils/src/errors.rs
@@ -1,5 +1,7 @@
 //! Errors and error specific types for universal use
 
+use serde::Serialize;
+
 use crate::types::MinorUnit;
 
 /// Custom Result
@@ -78,7 +80,7 @@ pub enum ValidationError {
 }
 
 /// Integrity check errors.
-#[derive(Debug, Clone, PartialEq, Default)]
+#[derive(Debug, Clone, PartialEq, Default, Serialize)]
 pub struct IntegrityCheckError {
     /// Field names for which integrity check failed!
     pub field_names: String,
diff --git a/crates/external_services/src/grpc_client.rs b/crates/external_services/src/grpc_client.rs
index a00b3168abc..5352c81c275 100644
--- a/crates/external_services/src/grpc_client.rs
+++ b/crates/external_services/src/grpc_client.rs
@@ -162,16 +162,23 @@ pub struct GrpcHeadersUcs {
     external_vault_proxy_metadata: Option<String>,
     /// Merchant Reference Id
     merchant_reference_id: Option<ucs_types::UcsReferenceId>,
+
+    request_id: Option<String>,
+
+    shadow_mode: Option<bool>,
 }
 
 /// Type aliase for GrpcHeaders builder in initial stage
-pub type GrpcHeadersUcsBuilderInitial = GrpcHeadersUcsBuilder<((String,), (), (), ())>;
+pub type GrpcHeadersUcsBuilderInitial =
+    GrpcHeadersUcsBuilder<((String,), (), (), (), (Option<String>,), (Option<bool>,))>;
 /// Type aliase for GrpcHeaders builder in intermediate stage
 pub type GrpcHeadersUcsBuilderFinal = GrpcHeadersUcsBuilder<(
     (String,),
     (LineageIds,),
     (Option<String>,),
     (Option<ucs_types::UcsReferenceId>,),
+    (Option<String>,),
+    (Option<bool>,),
 )>;
 
 /// struct to represent set of Lineage ids
diff --git a/crates/external_services/src/grpc_client/unified_connector_service.rs b/crates/external_services/src/grpc_client/unified_connector_service.rs
index 582ef85d729..b0799ae4ec4 100644
--- a/crates/external_services/src/grpc_client/unified_connector_service.rs
+++ b/crates/external_services/src/grpc_client/unified_connector_service.rs
@@ -456,6 +456,23 @@ pub fn build_unified_connector_service_grpc_headers(
         );
     };
 
+    if let Some(request_id) = grpc_headers.request_id {
+        metadata.append(
+            common_utils_consts::X_REQUEST_ID,
+            parse(common_utils_consts::X_REQUEST_ID, &request_id)?,
+        );
+    };
+
+    if let Some(shadow_mode) = grpc_headers.shadow_mode {
+        metadata.append(
+            common_utils_consts::X_UNIFIED_CONNECTOR_SERVICE_MODE,
+            parse(
+                common_utils_consts::X_UNIFIED_CONNECTOR_SERVICE_MODE,
+                &shadow_mode.to_string(),
+            )?,
+        );
+    }
+
     if let Err(err) = grpc_headers
         .tenant_id
         .parse()
diff --git a/crates/external_services/src/http_client/client.rs b/crates/external_services/src/http_client/client.rs
index 633634b6539..1f88f59507b 100644
--- a/crates/external_services/src/http_client/client.rs
+++ b/crates/external_services/src/http_client/client.rs
@@ -29,7 +29,8 @@ pub fn create_client(
         }
 
         logger::debug!("Creating HTTP client with mutual TLS (client cert + key)");
-        let client_builder = get_client_builder(proxy_config)?;
+        let client_builder =
+            apply_mitm_certificate(get_client_builder(proxy_config)?, proxy_config);
 
         let identity = create_identity_from_certificate_and_key(
             encoded_certificate.clone(),
@@ -56,7 +57,9 @@ pub fn create_client(
         let cert = reqwest::Certificate::from_pem(pem.as_bytes())
             .change_context(HttpClientError::ClientConstructionFailed)
             .attach_printable("Failed to parse CA certificate PEM block")?;
-        let client_builder = get_client_builder(proxy_config)?.add_root_certificate(cert);
+        let client_builder =
+            apply_mitm_certificate(get_client_builder(proxy_config)?, proxy_config)
+                .add_root_certificate(cert);
         return client_builder
             .use_rustls_tls()
             .build()
@@ -145,10 +148,32 @@ pub fn create_certificate(
         .change_context(HttpClientError::CertificateDecodeFailed)
 }
 
+fn apply_mitm_certificate(
+    mut client_builder: reqwest::ClientBuilder,
+    proxy_config: &Proxy,
+) -> reqwest::ClientBuilder {
+    if let Some(mitm_ca_cert) = &proxy_config.mitm_ca_certificate {
+        let pem = mitm_ca_cert.clone().expose().replace("\\r\\n", "\n");
+        match reqwest::Certificate::from_pem(pem.as_bytes()) {
+            Ok(cert) => {
+                logger::debug!("Successfully added MITM CA certificate");
+                client_builder = client_builder.add_root_certificate(cert);
+            }
+            Err(err) => {
+                logger::error!(
+                    "Failed to parse MITM CA certificate: {}, continuing without MITM support",
+                    err
+                );
+            }
+        }
+    }
+    client_builder
+}
+
 fn get_base_client(proxy_config: &Proxy) -> CustomResult<reqwest::Client, HttpClientError> {
     Ok(DEFAULT_CLIENT
         .get_or_try_init(|| {
-            get_client_builder(proxy_config)?
+            apply_mitm_certificate(get_client_builder(proxy_config)?, proxy_config)
                 .build()
                 .change_context(HttpClientError::ClientConstructionFailed)
                 .attach_printable("Failed to construct base client")
diff --git a/crates/hyperswitch_domain_models/src/mandates.rs b/crates/hyperswitch_domain_models/src/mandates.rs
index b80e50b5b46..ecfd7f34694 100644
--- a/crates/hyperswitch_domain_models/src/mandates.rs
+++ b/crates/hyperswitch_domain_models/src/mandates.rs
@@ -56,7 +56,7 @@ pub struct MandateAmountData {
 
 // The fields on this struct are optional, as we want to allow the merchant to provide partial
 // information about creating mandates
-#[derive(Default, Eq, PartialEq, Debug, Clone)]
+#[derive(Default, Eq, PartialEq, Debug, Clone, serde::Serialize)]
 pub struct MandateData {
     /// A way to update the mandate's payment method details
     pub update_mandate_id: Option<String>,
diff --git a/crates/hyperswitch_domain_models/src/payment_address.rs b/crates/hyperswitch_domain_models/src/payment_address.rs
index 176bef0c88b..4240009931c 100644
--- a/crates/hyperswitch_domain_models/src/payment_address.rs
+++ b/crates/hyperswitch_domain_models/src/payment_address.rs
@@ -1,6 +1,6 @@
 use crate::address::Address;
 
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Default, Debug, serde::Serialize)]
 pub struct PaymentAddress {
     shipping: Option<Address>,
     billing: Option<Address>,
diff --git a/crates/hyperswitch_domain_models/src/payment_method_data.rs b/crates/hyperswitch_domain_models/src/payment_method_data.rs
index 241523db118..80b2fd89d31 100644
--- a/crates/hyperswitch_domain_models/src/payment_method_data.rs
+++ b/crates/hyperswitch_domain_models/src/payment_method_data.rs
@@ -51,7 +51,7 @@ pub enum ExternalVaultPaymentMethodData {
     VaultToken(VaultToken),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
 pub enum ApplePayFlow {
     Simplified(api_models::payments::PaymentProcessingDetails),
     Manual,
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index a01efdf0749..80c671c0978 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -849,7 +849,7 @@ impl PaymentIntent {
 }
 
 #[cfg(feature = "v1")]
-#[derive(Default, Debug, Clone)]
+#[derive(Default, Debug, Clone, serde::Serialize)]
 pub struct HeaderPayload {
     pub payment_confirm_source: Option<common_enums::PaymentSource>,
     pub client_source: Option<String>,
@@ -879,7 +879,7 @@ pub struct ClickToPayMetaData {
 
 // TODO: uncomment fields as necessary
 #[cfg(feature = "v2")]
-#[derive(Default, Debug, Clone)]
+#[derive(Default, Debug, Clone, serde::Serialize)]
 pub struct HeaderPayload {
     /// The source with which the payment is confirmed.
     pub payment_confirm_source: Option<common_enums::PaymentSource>,
diff --git a/crates/hyperswitch_domain_models/src/router_data.rs b/crates/hyperswitch_domain_models/src/router_data.rs
index 5f942e83b3d..d0bffe88882 100644
--- a/crates/hyperswitch_domain_models/src/router_data.rs
+++ b/crates/hyperswitch_domain_models/src/router_data.rs
@@ -9,6 +9,7 @@ use common_utils::{
 };
 use error_stack::ResultExt;
 use masking::{ExposeInterface, Secret};
+use serde::{Deserialize, Serialize};
 
 use crate::{
     network_tokenization::NetworkTokenNumber, payment_address::PaymentAddress, payment_method_data,
@@ -23,7 +24,7 @@ use crate::{
     router_flow_types, router_request_types, router_response_types,
 };
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct RouterData<Flow, Request, Response> {
     pub flow: PhantomData<Flow>,
     pub merchant_id: id_type::MerchantId,
@@ -118,7 +119,7 @@ pub struct RouterData<Flow, Request, Response> {
     pub is_payment_id_from_merchant: Option<bool>,
 }
 
-#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct L2L3Data {
     pub order_date: Option<time::PrimitiveDateTime>,
     pub tax_status: Option<common_enums::TaxStatus>,
@@ -140,7 +141,7 @@ pub struct L2L3Data {
 }
 
 // Different patterns of authentication.
-#[derive(Default, Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Default, Debug, Clone, Deserialize, Serialize)]
 #[serde(tag = "auth_type")]
 pub enum ConnectorAuthType {
     TemporaryAuth,
@@ -263,19 +264,19 @@ impl ConnectorAuthType {
     }
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct AccessTokenAuthenticationResponse {
     pub code: Secret<String>,
     pub expires: i64,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
+#[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct AccessToken {
     pub token: Secret<String>,
     pub expires: i64,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub enum PaymentMethodToken {
     Token(Secret<String>),
     ApplePayDecrypt(Box<common_payment_types::ApplePayPredecryptData>),
@@ -283,7 +284,7 @@ pub enum PaymentMethodToken {
     PazeDecrypt(Box<PazeDecryptedData>),
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ApplePayPredecryptDataInternal {
     pub application_primary_account_number: cards::CardNumber,
@@ -295,7 +296,7 @@ pub struct ApplePayPredecryptDataInternal {
     pub payment_data: ApplePayCryptogramDataInternal,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ApplePayCryptogramDataInternal {
     pub online_payment_cryptogram: Secret<String>,
@@ -365,7 +366,7 @@ impl ApplePayPredecryptDataInternal {
     }
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GooglePayPredecryptDataInternal {
     pub message_expiration: String,
@@ -375,7 +376,7 @@ pub struct GooglePayPredecryptDataInternal {
     pub payment_method_details: GooglePayPaymentMethodDetails,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GooglePayPaymentMethodDetails {
     pub auth_method: common_enums::enums::GooglePayAuthMethod,
@@ -387,7 +388,7 @@ pub struct GooglePayPaymentMethodDetails {
     pub card_network: Option<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazeDecryptedData {
     pub client_id: Secret<String>,
@@ -400,7 +401,7 @@ pub struct PazeDecryptedData {
     pub eci: Option<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazeToken {
     pub payment_token: NetworkTokenNumber,
@@ -409,7 +410,7 @@ pub struct PazeToken {
     pub payment_account_reference: Secret<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazeDynamicData {
     pub dynamic_data_value: Option<Secret<String>>,
@@ -417,7 +418,7 @@ pub struct PazeDynamicData {
     pub dynamic_data_expiration: Option<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazeAddress {
     pub name: Option<Secret<String>>,
@@ -430,7 +431,7 @@ pub struct PazeAddress {
     pub country_code: Option<common_enums::enums::CountryAlpha2>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazeConsumer {
     // This is consumer data not customer data.
@@ -443,14 +444,14 @@ pub struct PazeConsumer {
     pub language_code: Option<String>,
 }
 
-#[derive(Debug, Clone, serde::Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PazePhoneNumber {
     pub country_code: Secret<String>,
     pub phone_number: Secret<String>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize)]
 pub struct RecurringMandatePaymentData {
     pub payment_method_type: Option<common_enums::enums::PaymentMethodType>, //required for making recurring payment using saved payment method through stripe
     pub original_payment_authorized_amount: Option<i64>,
@@ -458,13 +459,13 @@ pub struct RecurringMandatePaymentData {
     pub mandate_metadata: Option<common_utils::pii::SecretSerdeValue>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentMethodBalance {
     pub amount: MinorUnit,
     pub currency: common_enums::enums::Currency,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ConnectorResponseData {
     pub additional_payment_method_data: Option<AdditionalPaymentMethodConnectorResponse>,
     extended_authorization_response_data: Option<ExtendedAuthorizationResponseData>,
@@ -504,7 +505,7 @@ impl ConnectorResponseData {
     }
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum AdditionalPaymentMethodConnectorResponse {
     Card {
         /// Details regarding the authentication details of the connector, if this is a 3ds payment.
@@ -521,19 +522,19 @@ pub enum AdditionalPaymentMethodConnectorResponse {
     },
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ExtendedAuthorizationResponseData {
     pub extended_authentication_applied:
         Option<primitive_wrappers::ExtendedAuthorizationAppliedBool>,
     pub capture_before: Option<time::PrimitiveDateTime>,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KlarnaSdkResponse {
     pub payment_type: Option<String>,
 }
 
-#[derive(Clone, Debug, serde::Serialize)]
+#[derive(Clone, Debug, Serialize)]
 pub struct ErrorResponse {
     pub code: String,
     pub message: String,
diff --git a/crates/hyperswitch_domain_models/src/router_request_types.rs b/crates/hyperswitch_domain_models/src/router_request_types.rs
index 87cf479339c..995fd9f2f38 100644
--- a/crates/hyperswitch_domain_models/src/router_request_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_request_types.rs
@@ -23,7 +23,7 @@ use crate::{
     router_flow_types as flows, router_response_types as response_types,
     vault::PaymentMethodVaultingData,
 };
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsAuthorizeData {
     pub payment_method_data: PaymentMethodData,
     /// total amount (original_amount + surcharge_amount + tax_on_surcharge_amount)
@@ -92,7 +92,7 @@ pub struct PaymentsAuthorizeData {
     pub mit_category: Option<common_enums::MitCategory>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ExternalVaultProxyPaymentsData {
     pub payment_method_data: ExternalVaultPaymentMethodData,
     /// total amount (original_amount + surcharge_amount + tax_on_surcharge_amount)
@@ -191,7 +191,7 @@ impl
         })
     }
 }
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsPostSessionTokensData {
     // amount here would include amount, surcharge_amount and shipping_cost
     pub amount: MinorUnit,
@@ -208,13 +208,13 @@ pub struct PaymentsPostSessionTokensData {
     pub router_return_url: Option<String>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsUpdateMetadataData {
     pub metadata: pii::SecretSerdeValue,
     pub connector_transaction_id: String,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize)]
 pub struct AuthoriseIntegrityObject {
     /// Authorise amount
     pub amount: MinorUnit,
@@ -222,7 +222,7 @@ pub struct AuthoriseIntegrityObject {
     pub currency: storage_enums::Currency,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize)]
 pub struct SyncIntegrityObject {
     /// Sync amount
     pub amount: Option<MinorUnit>,
@@ -230,7 +230,7 @@ pub struct SyncIntegrityObject {
     pub currency: Option<storage_enums::Currency>,
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize)]
 pub struct PaymentsCaptureData {
     pub amount_to_capture: i64,
     pub currency: storage_enums::Currency,
@@ -250,7 +250,7 @@ pub struct PaymentsCaptureData {
     pub webhook_url: Option<String>,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize)]
 pub struct CaptureIntegrityObject {
     /// capture amount
     pub capture_amount: Option<MinorUnit>,
@@ -258,7 +258,7 @@ pub struct CaptureIntegrityObject {
     pub currency: storage_enums::Currency,
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize)]
 pub struct PaymentsIncrementalAuthorizationData {
     pub total_amount: i64,
     pub additional_amount: i64,
@@ -268,13 +268,13 @@ pub struct PaymentsIncrementalAuthorizationData {
     pub connector_meta: Option<serde_json::Value>,
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize)]
 pub struct MultipleCaptureRequestData {
     pub capture_sequence: i16,
     pub capture_reference: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct AuthorizeSessionTokenData {
     pub amount_to_capture: Option<i64>,
     pub currency: storage_enums::Currency,
@@ -282,7 +282,7 @@ pub struct AuthorizeSessionTokenData {
     pub amount: Option<i64>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ConnectorCustomerData {
     pub description: Option<String>,
     pub email: Option<pii::Email>,
@@ -407,7 +407,7 @@ impl TryFrom<&RouterData<flows::Session, PaymentsSessionData, response_types::Pa
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentMethodTokenizationData {
     pub payment_method_data: PaymentMethodData,
     pub browser_info: Option<BrowserInformation>,
@@ -513,7 +513,7 @@ impl TryFrom<ExternalVaultProxyPaymentsData> for PaymentMethodTokenizationData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct CreateOrderRequestData {
     pub minor_amount: MinorUnit,
     pub currency: storage_enums::Currency,
@@ -541,7 +541,7 @@ impl TryFrom<ExternalVaultProxyPaymentsData> for CreateOrderRequestData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsPreProcessingData {
     pub payment_method_data: Option<PaymentMethodData>,
     pub amount: Option<i64>,
@@ -570,7 +570,7 @@ pub struct PaymentsPreProcessingData {
     pub is_stored_credential: Option<bool>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct GiftCardBalanceCheckRequestData {
     pub payment_method_data: PaymentMethodData,
     pub currency: Option<storage_enums::Currency>,
@@ -610,7 +610,7 @@ impl TryFrom<PaymentsAuthorizeData> for PaymentsPreProcessingData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsPreAuthenticateData {
     pub payment_method_data: Option<PaymentMethodData>,
     pub amount: Option<i64>,
@@ -649,7 +649,7 @@ impl TryFrom<PaymentsAuthorizeData> for PaymentsPreAuthenticateData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsAuthenticateData {
     pub payment_method_data: Option<PaymentMethodData>,
     pub amount: Option<i64>,
@@ -688,7 +688,7 @@ impl TryFrom<PaymentsAuthorizeData> for PaymentsAuthenticateData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsPostAuthenticateData {
     pub payment_method_data: Option<PaymentMethodData>,
     pub amount: Option<i64>,
@@ -760,7 +760,7 @@ impl TryFrom<CompleteAuthorizeData> for PaymentsPreProcessingData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsPostProcessingData {
     pub payment_method_data: PaymentMethodData,
     pub customer_id: Option<id_type::CustomerId>,
@@ -798,7 +798,7 @@ impl<F> TryFrom<RouterData<F, PaymentsAuthorizeData, response_types::PaymentsRes
         })
     }
 }
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct CompleteAuthorizeData {
     pub payment_method_data: Option<PaymentMethodData>,
     pub amount: i64,
@@ -827,13 +827,13 @@ pub struct CompleteAuthorizeData {
     pub is_stored_credential: Option<bool>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct CompleteAuthorizeRedirectResponse {
     pub params: Option<Secret<String>>,
     pub payload: Option<pii::SecretSerdeValue>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize)]
 pub struct PaymentsSyncData {
     //TODO : add fields based on the connector requirements
     pub connector_transaction_id: ResponseId,
@@ -852,14 +852,14 @@ pub struct PaymentsSyncData {
     pub setup_future_usage: Option<storage_enums::FutureUsage>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize)]
 pub enum SyncRequestType {
     MultipleCaptureSync(Vec<String>),
     #[default]
     SinglePaymentSync,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize)]
 pub struct PaymentsCancelData {
     pub amount: Option<i64>,
     pub currency: Option<storage_enums::Currency>,
@@ -876,7 +876,7 @@ pub struct PaymentsCancelData {
     pub capture_method: Option<storage_enums::CaptureMethod>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Serialize)]
 pub struct PaymentsCancelPostCaptureData {
     pub currency: Option<storage_enums::Currency>,
     pub connector_transaction_id: String,
@@ -898,7 +898,7 @@ pub struct PaymentsApproveData {
     pub currency: Option<storage_enums::Currency>,
 }
 
-#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
+#[derive(Clone, Debug, Default, Serialize, serde::Deserialize)]
 pub struct BrowserInformation {
     pub color_depth: Option<u8>,
     pub java_enabled: Option<bool>,
@@ -984,7 +984,7 @@ impl ResponseId {
     }
 }
 
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, Serialize)]
 pub struct SurchargeDetails {
     /// original_amount
     pub original_amount: MinorUnit,
@@ -1049,7 +1049,7 @@ impl
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct AuthenticationData {
     pub eci: Option<String>,
     pub cavv: Secret<String>,
@@ -1129,18 +1129,18 @@ pub struct ChargeRefunds {
     pub options: ChargeRefundsOptions,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, Serialize)]
 pub enum ChargeRefundsOptions {
     Destination(DestinationChargeRefund),
     Direct(DirectChargeRefund),
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, Serialize)]
 pub struct DirectChargeRefund {
     pub revert_platform_fee: bool,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, Serialize)]
 pub struct DestinationChargeRefund {
     pub revert_platform_fee: bool,
     pub revert_transfer: bool,
@@ -1310,7 +1310,7 @@ pub struct RetrieveFileRequestData {
 }
 
 #[serde_as]
-#[derive(Clone, Debug, serde::Serialize)]
+#[derive(Clone, Debug, Serialize)]
 pub struct UploadFileRequestData {
     pub file_key: String,
     #[serde(skip)]
@@ -1366,7 +1366,7 @@ pub struct MandateRevokeRequestData {
     pub connector_mandate_id: Option<String>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PaymentsSessionData {
     pub amount: i64,
     pub currency: common_enums::Currency,
@@ -1395,7 +1395,7 @@ pub struct PaymentsTaxCalculationData {
     pub shipping_address: address::Address,
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize)]
 pub struct SdkPaymentsSessionUpdateData {
     pub order_tax_amount: MinorUnit,
     // amount here would include amount, surcharge_amount, order_tax_amount and shipping_cost
@@ -1407,7 +1407,7 @@ pub struct SdkPaymentsSessionUpdateData {
     pub shipping_cost: Option<MinorUnit>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct SetupMandateRequestData {
     pub currency: storage_enums::Currency,
     pub payment_method_data: PaymentMethodData,
diff --git a/crates/hyperswitch_domain_models/src/router_response_types.rs b/crates/hyperswitch_domain_models/src/router_response_types.rs
index 3481a61dfb0..5b50037b617 100644
--- a/crates/hyperswitch_domain_models/src/router_response_types.rs
+++ b/crates/hyperswitch_domain_models/src/router_response_types.rs
@@ -10,6 +10,7 @@ pub use disputes::{
     AcceptDisputeResponse, DefendDisputeResponse, DisputeSyncResponse, FetchDisputesResponse,
     SubmitEvidenceResponse,
 };
+use serde::Serialize;
 
 use crate::{
     errors::api_error_response::ApiErrorResponse,
@@ -24,7 +25,7 @@ pub struct RefundsResponseData {
     // pub amount_received: Option<i32>, // Calculation for amount received not in place yet
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ConnectorCustomerResponseData {
     pub connector_customer_id: String,
     pub name: Option<String>,
@@ -51,7 +52,7 @@ impl ConnectorCustomerResponseData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub enum PaymentsResponseData {
     TransactionResponse {
         resource_id: ResponseId,
@@ -123,7 +124,7 @@ pub struct TaxCalculationResponseData {
     pub order_tax_amount: MinorUnit,
 }
 
-#[derive(serde::Serialize, Debug, Clone)]
+#[derive(Serialize, Debug, Clone)]
 pub struct MandateReference {
     pub connector_mandate_id: Option<String>,
     pub payment_method_id: Option<String>,
@@ -131,7 +132,7 @@ pub struct MandateReference {
     pub connector_mandate_request_reference_id: Option<String>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub enum CaptureSyncResponse {
     Success {
         resource_id: ResponseId,
@@ -282,13 +283,13 @@ impl PaymentsResponseData {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub enum PreprocessingResponseId {
     PreProcessingId(String),
     ConnectorTransactionId(String),
 }
 
-#[derive(Debug, Eq, PartialEq, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, serde::Deserialize)]
 pub enum RedirectForm {
     Form {
         endpoint: String,
diff --git a/crates/hyperswitch_interfaces/src/types.rs b/crates/hyperswitch_interfaces/src/types.rs
index 5003fdba1c7..6b9203a7363 100644
--- a/crates/hyperswitch_interfaces/src/types.rs
+++ b/crates/hyperswitch_interfaces/src/types.rs
@@ -399,6 +399,12 @@ pub struct Proxy {
 
     /// A comma-separated list of hosts that should bypass the proxy.
     pub bypass_proxy_hosts: Option<String>,
+
+    /// The CA certificate used for man-in-the-middle (MITM) proxying, if enabled.
+    pub mitm_ca_certificate: Option<masking::Secret<String>>,
+
+    /// Whether man-in-the-middle (MITM) proxying is enabled.
+    pub mitm_enabled: Option<bool>,
 }
 
 impl Default for Proxy {
@@ -408,6 +414,8 @@ impl Default for Proxy {
             https_url: Default::default(),
             idle_pool_connection_timeout: Some(90),
             bypass_proxy_hosts: Default::default(),
+            mitm_ca_certificate: None,
+            mitm_enabled: None,
         }
     }
 }
diff --git a/crates/router/src/configs/secrets_transformers.rs b/crates/router/src/configs/secrets_transformers.rs
index 37b551081ce..0f59774c6d6 100644
--- a/crates/router/src/configs/secrets_transformers.rs
+++ b/crates/router/src/configs/secrets_transformers.rs
@@ -591,5 +591,6 @@ pub(crate) async fn fetch_raw_secrets(
         proxy_status_mapping: conf.proxy_status_mapping,
         internal_services: conf.internal_services,
         superposition,
+        comparison_service: conf.comparison_service,
     }
 }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index 28b21cc7e0c..743c308192d 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -7,7 +7,11 @@ use std::{
 #[cfg(feature = "olap")]
 use analytics::{opensearch::OpenSearchConfig, ReportConfig};
 use api_models::enums;
-use common_utils::{ext_traits::ConfigExt, id_type, types::user::EmailThemeConfig};
+use common_utils::{
+    ext_traits::ConfigExt,
+    id_type,
+    types::{user::EmailThemeConfig, Url},
+};
 use config::{Environment, File};
 use error_stack::ResultExt;
 #[cfg(feature = "email")]
@@ -172,6 +176,7 @@ pub struct Settings<S: SecretState> {
     pub superposition: SecretStateContainer<SuperpositionClientConfig, S>,
     pub proxy_status_mapping: ProxyStatusMapping,
     pub internal_services: InternalServicesConfig,
+    pub comparison_service: Option<ComparisonServiceConfig>,
 }
 
 #[derive(Debug, Deserialize, Clone, Default)]
@@ -200,6 +205,13 @@ pub struct CloneConnectorAllowlistConfig {
     pub connector_names: HashSet<enums::Connector>,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+pub struct ComparisonServiceConfig {
+    pub url: Url,
+    pub enabled: bool,
+    pub timeout_secs: Option<u64>,
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct Platform {
     pub enabled: bool,
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 0b9f7de40d4..e802064730a 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -36,7 +36,7 @@ use api_models::{
     mandates::RecurringDetails,
     payments::{self as payments_api},
 };
-pub use common_enums::enums::{CallConnectorAction, GatewaySystem};
+pub use common_enums::enums::{CallConnectorAction, ExecutionMode, GatewaySystem};
 use common_types::payments as common_payments_types;
 use common_utils::{
     ext_traits::{AsyncExt, StringExt},
@@ -96,8 +96,12 @@ use self::{
 #[cfg(feature = "v1")]
 use super::unified_connector_service::update_gateway_system_in_feature_metadata;
 use super::{
-    errors::StorageErrorExt, payment_methods::surcharge_decision_configs, routing::TransactionData,
-    unified_connector_service::should_call_unified_connector_service,
+    errors::StorageErrorExt,
+    payment_methods::surcharge_decision_configs,
+    routing::TransactionData,
+    unified_connector_service::{
+        send_comparison_data, should_call_unified_connector_service, ComparisonData,
+    },
 };
 #[cfg(feature = "v1")]
 use crate::core::debit_routing;
@@ -456,7 +460,7 @@ where
 
     // To perform router related operation for PaymentResponse
     PaymentResponse: Operation<F, FData, Data = D>,
-    FData: Send + Sync + Clone,
+    FData: Send + Sync + Clone + serde::Serialize,
 {
     let operation: BoxedOperation<'_, F, Req, D> = Box::new(operation);
 
@@ -534,7 +538,7 @@ where
 #[cfg(feature = "v1")]
 #[allow(clippy::too_many_arguments, clippy::type_complexity)]
 #[instrument(skip_all, fields(payment_id, merchant_id))]
-pub async fn payments_operation_core<F, Req, Op, FData, D>(
+pub async fn payments_operation_core<'a, F, Req, Op, FData, D>(
     state: &SessionState,
     req_state: ReqState,
     merchant_context: &domain::MerchantContext,
@@ -547,7 +551,7 @@ pub async fn payments_operation_core<F, Req, Op, FData, D>(
     header_payload: HeaderPayload,
 ) -> RouterResult<(D, Req, Option<domain::Customer>, Option<u16>, Option<u128>)>
 where
-    F: Send + Clone + Sync,
+    F: Send + Clone + Sync + 'static + 'a,
     Req: Authenticate + Clone,
     Op: Operation<F, Req, Data = D> + Send + Sync,
     D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
@@ -562,7 +566,7 @@ where
 
     // To perform router related operation for PaymentResponse
     PaymentResponse: Operation<F, FData, Data = D>,
-    FData: Send + Sync + Clone + router_types::Capturable,
+    FData: Send + Sync + Clone + router_types::Capturable + 'static + 'a + serde::Serialize,
 {
     let operation: BoxedOperation<'_, F, Req, D> = Box::new(operation);
 
@@ -1242,8 +1246,9 @@ where
     RouterData<F, FData, router_types::PaymentsResponseData>: Feature<F, FData>,
 
     // To construct connector flow specific api
-    dyn api::Connector:
-        services::api::ConnectorIntegration<F, FData, router_types::PaymentsResponseData>,
+    dyn api::Connector: services::api::ConnectorIntegration<F, FData, router_types::PaymentsResponseData>
+        + Send
+        + Sync,
 
     // To perform router related operation for PaymentResponse
     PaymentResponse: Operation<F, FData, Data = D>,
@@ -2087,7 +2092,7 @@ pub async fn call_surcharge_decision_management_for_session_flow(
 
 #[cfg(feature = "v1")]
 #[allow(clippy::too_many_arguments)]
-pub async fn payments_core<F, Res, Req, Op, FData, D>(
+pub async fn payments_core<'a, F, Res, Req, Op, FData, D>(
     state: SessionState,
     req_state: ReqState,
     merchant_context: domain::MerchantContext,
@@ -2100,8 +2105,8 @@ pub async fn payments_core<F, Res, Req, Op, FData, D>(
     header_payload: HeaderPayload,
 ) -> RouterResponse<Res>
 where
-    F: Send + Clone + Sync,
-    FData: Send + Sync + Clone + router_types::Capturable,
+    F: Send + Clone + Sync + 'static + 'a,
+    FData: Send + Sync + Clone + router_types::Capturable + 'static + 'a + serde::Serialize,
     Op: Operation<F, Req, Data = D> + Send + Sync + Clone,
     Req: Debug + Authenticate + Clone,
     D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
@@ -2109,7 +2114,6 @@ where
     // To create connector flow specific interface data
     D: ConstructFlowSpecificData<F, FData, router_types::PaymentsResponseData>,
     RouterData<F, FData, router_types::PaymentsResponseData>: Feature<F, FData>,
-
     // To construct connector flow specific api
     dyn api::Connector:
         services::api::ConnectorIntegration<F, FData, router_types::PaymentsResponseData>,
@@ -2671,7 +2675,7 @@ pub async fn payments_core<F, Res, Req, Op, FData, D>(
 where
     F: Send + Clone + Sync,
     Req: Send + Sync + Authenticate,
-    FData: Send + Sync + Clone,
+    FData: Send + Sync + Clone + serde::Serialize,
     Op: Operation<F, Req, Data = D> + ValidateStatusForOperation + Send + Sync + Clone,
     Req: Debug,
     D: OperationSessionGetters<F>
@@ -3903,8 +3907,9 @@ where
     D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
     RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
     // To construct connector flow specific api
-    dyn api::Connector:
-        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+    dyn api::Connector: services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>
+        + Send
+        + Sync,
 {
     let add_access_token_result = router_data
         .add_access_token(
@@ -4081,12 +4086,13 @@ pub async fn call_connector_service_prerequisites<F, RouterDReq, ApiRequest, D>(
 )>
 where
     F: Send + Clone + Sync,
-    RouterDReq: Send + Sync,
+    RouterDReq: Send + Clone + Sync,
 
     // To create connector flow specific interface data
     D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
     D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
-    RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone,
     // To construct connector flow specific api
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
@@ -4220,150 +4226,497 @@ where
 #[cfg(feature = "v1")]
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
-pub async fn decide_unified_connector_service_call<F, RouterDReq, ApiRequest, D>(
-    state: &SessionState,
+pub async fn decide_unified_connector_service_call<'a, F, RouterDReq, ApiRequest, D>(
+    state: &'a SessionState,
     req_state: ReqState,
-    merchant_context: &domain::MerchantContext,
+    merchant_context: &'a domain::MerchantContext,
     connector: api::ConnectorData,
-    operation: &BoxedOperation<'_, F, ApiRequest, D>,
-    payment_data: &mut D,
+    operation: &'a BoxedOperation<'a, F, ApiRequest, D>,
+    payment_data: &'a mut D,
     customer: &Option<domain::Customer>,
     call_connector_action: CallConnectorAction,
-    validate_result: &operations::ValidateResult,
+    validate_result: &'a operations::ValidateResult,
     schedule_time: Option<time::PrimitiveDateTime>,
     header_payload: HeaderPayload,
     frm_suggestion: Option<storage_enums::FrmSuggestion>,
-    business_profile: &domain::Profile,
+    business_profile: &'a domain::Profile,
     is_retry_payment: bool,
     all_keys_required: Option<bool>,
     merchant_connector_account: helpers::MerchantConnectorAccountType,
-    mut router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
     tokenization_action: TokenizationAction,
 ) -> RouterResult<(
     RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
     helpers::MerchantConnectorAccountType,
 )>
 where
-    F: Send + Clone + Sync,
-    RouterDReq: Send + Sync,
+    F: Send + Clone + Sync + 'static + 'a,
+    RouterDReq: Send + Sync + Clone + 'static + 'a + serde::Serialize,
 
     // To create connector flow specific interface data
     D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
     D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
-    RouterData<F, RouterDReq, router_types::PaymentsResponseData>: Feature<F, RouterDReq> + Send,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone + serde::Serialize,
     // To construct connector flow specific api
     dyn api::Connector:
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
-    record_time_taken_with(|| async {
-        if !matches!(
-            call_connector_action,
-            CallConnectorAction::UCSHandleResponse(_)
-        ) && !matches!(
-            call_connector_action,
-            CallConnectorAction::HandleResponse(_),
-        ) && should_call_unified_connector_service(
-            state,
-            merchant_context,
-            &router_data,
-            Some(payment_data),
-        )
-        .await?
-        {
-            router_env::logger::info!(
-                "Processing payment through UCS gateway system - payment_id={}, attempt_id={}",
-                payment_data
-                    .get_payment_intent()
-                    .payment_id
-                    .get_string_repr(),
-                payment_data.get_payment_attempt().attempt_id
-            );
-            if should_add_task_to_process_tracker(payment_data) {
-                operation
-                    .to_domain()?
-                    .add_task_to_process_tracker(
-                        state,
-                        payment_data.get_payment_attempt(),
-                        validate_result.requeue,
-                        schedule_time,
-                    )
-                    .await
-                    .map_err(|error| logger::error!(process_tracker_error=?error))
-                    .ok();
-            }
+    let execution_path = should_call_unified_connector_service(
+        state,
+        merchant_context,
+        &router_data,
+        Some(payment_data),
+    )
+    .await?;
 
-            // Update feature metadata to track UCS usage for stickiness
-            update_gateway_system_in_feature_metadata(
-                payment_data,
-                GatewaySystem::UnifiedConnectorService,
-            )?;
+    let is_handle_response_action = matches!(
+        call_connector_action,
+        CallConnectorAction::UCSHandleResponse(_) | CallConnectorAction::HandleResponse(_)
+    );
 
-            (_, *payment_data) = operation
-                .to_update_tracker()?
-                .update_trackers(
+    record_time_taken_with(|| async {
+        match (execution_path, is_handle_response_action) {
+            // Process through UCS when system is UCS and not handling response
+            (GatewaySystem::UnifiedConnectorService, false) => {
+                process_through_ucs(
                     state,
                     req_state,
-                    payment_data.clone(),
-                    customer.clone(),
-                    merchant_context.get_merchant_account().storage_scheme,
-                    None,
-                    merchant_context.get_merchant_key_store(),
+                    merchant_context,
+                    operation,
+                    payment_data,
+                    customer,
+                    validate_result,
+                    schedule_time,
+                    header_payload,
                     frm_suggestion,
-                    header_payload.clone(),
+                    business_profile,
+                    merchant_connector_account,
+                    router_data,
                 )
-                .await?;
-            let lineage_ids = grpc_client::LineageIds::new(
-                business_profile.merchant_id.clone(),
-                business_profile.get_id().clone(),
-            );
-            router_data
-                .call_unified_connector_service(
+                .await
+            }
+
+            // Process through Direct gateway
+            (GatewaySystem::Direct, _) | (GatewaySystem::UnifiedConnectorService, true) => {
+                process_through_direct(
                     state,
-                    &header_payload,
-                    lineage_ids,
-                    merchant_connector_account.clone(),
+                    req_state,
                     merchant_context,
+                    connector,
+                    operation,
+                    payment_data,
+                    customer,
+                    call_connector_action,
+                    validate_result,
+                    schedule_time,
+                    header_payload,
+                    frm_suggestion,
+                    business_profile,
+                    is_retry_payment,
+                    all_keys_required,
+                    merchant_connector_account,
+                    router_data,
+                    tokenization_action,
                 )
-                .await?;
+                .await
+            }
 
-            Ok((router_data, merchant_connector_account))
-        } else {
-            router_env::logger::info!(
-                "Processing payment through Direct gateway system - payment_id={}, attempt_id={}",
-                payment_data
-                    .get_payment_intent()
-                    .payment_id
-                    .get_string_repr(),
-                payment_data.get_payment_attempt().attempt_id
-            );
+            // Process through Direct with Shadow UCS
+            (GatewaySystem::ShadowUnifiedConnectorService, _) => {
+                process_through_direct_with_shadow_unified_connector_service(
+                    state,
+                    req_state,
+                    merchant_context,
+                    connector,
+                    operation,
+                    payment_data,
+                    customer,
+                    call_connector_action,
+                    validate_result,
+                    schedule_time,
+                    header_payload,
+                    frm_suggestion,
+                    business_profile,
+                    is_retry_payment,
+                    all_keys_required,
+                    merchant_connector_account,
+                    router_data,
+                    tokenization_action,
+                )
+                .await
+            }
+        }
+    })
+    .await
+}
 
-            // Update feature metadata to track Direct routing usage for stickiness
-            update_gateway_system_in_feature_metadata(payment_data, GatewaySystem::Direct)?;
+#[cfg(feature = "v1")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+// Helper function to process through UCS
+async fn process_through_ucs<'a, F, RouterDReq, ApiRequest, D>(
+    state: &'a SessionState,
+    req_state: ReqState,
+    merchant_context: &'a domain::MerchantContext,
+    operation: &'a BoxedOperation<'a, F, ApiRequest, D>,
+    payment_data: &'a mut D,
+    customer: &Option<domain::Customer>,
+    validate_result: &'a operations::ValidateResult,
+    schedule_time: Option<time::PrimitiveDateTime>,
+    header_payload: HeaderPayload,
+    frm_suggestion: Option<storage_enums::FrmSuggestion>,
+    business_profile: &'a domain::Profile,
+    merchant_connector_account: helpers::MerchantConnectorAccountType,
+    mut router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+) -> RouterResult<(
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    helpers::MerchantConnectorAccountType,
+)>
+where
+    F: Send + Clone + Sync + 'static + 'a,
+    RouterDReq: Send + Sync + Clone + 'static + 'a + serde::Serialize,
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone + serde::Serialize,
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    router_env::logger::info!(
+        "Processing payment through UCS gateway system - payment_id={}, attempt_id={}",
+        payment_data
+            .get_payment_intent()
+            .payment_id
+            .get_string_repr(),
+        payment_data.get_payment_attempt().attempt_id
+    );
 
-            call_connector_service(
+    // Add task to process tracker if needed
+    if should_add_task_to_process_tracker(payment_data) {
+        operation
+            .to_domain()?
+            .add_task_to_process_tracker(
                 state,
-                req_state,
-                merchant_context,
-                connector,
-                operation,
-                payment_data,
-                customer,
-                call_connector_action,
-                validate_result,
+                payment_data.get_payment_attempt(),
+                validate_result.requeue,
                 schedule_time,
-                header_payload,
-                frm_suggestion,
-                business_profile,
-                is_retry_payment,
-                all_keys_required,
-                merchant_connector_account,
-                router_data,
-                tokenization_action,
             )
             .await
-        }
-    })
+            .map_err(|error| logger::error!(process_tracker_error=?error))
+            .ok();
+    }
+
+    // Update feature metadata to track UCS usage for stickiness
+    update_gateway_system_in_feature_metadata(
+        payment_data,
+        GatewaySystem::UnifiedConnectorService,
+    )?;
+
+    // Update trackers
+    (_, *payment_data) = operation
+        .to_update_tracker()?
+        .update_trackers(
+            state,
+            req_state,
+            payment_data.clone(),
+            customer.clone(),
+            merchant_context.get_merchant_account().storage_scheme,
+            None,
+            merchant_context.get_merchant_key_store(),
+            frm_suggestion,
+            header_payload.clone(),
+        )
+        .await?;
+
+    // Call UCS
+    let lineage_ids = grpc_client::LineageIds::new(
+        business_profile.merchant_id.clone(),
+        business_profile.get_id().clone(),
+    );
+
+    router_data
+        .call_unified_connector_service(
+            state,
+            &header_payload,
+            lineage_ids,
+            merchant_connector_account.clone(),
+            merchant_context,
+            ExecutionMode::Primary, // UCS is called in primary mode
+        )
+        .await?;
+
+    Ok((router_data, merchant_connector_account))
+}
+
+#[cfg(feature = "v1")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+// Helper function to process through Direct gateway
+async fn process_through_direct<'a, F, RouterDReq, ApiRequest, D>(
+    state: &'a SessionState,
+    req_state: ReqState,
+    merchant_context: &'a domain::MerchantContext,
+    connector: api::ConnectorData,
+    operation: &'a BoxedOperation<'a, F, ApiRequest, D>,
+    payment_data: &'a mut D,
+    customer: &Option<domain::Customer>,
+    call_connector_action: CallConnectorAction,
+    validate_result: &'a operations::ValidateResult,
+    schedule_time: Option<time::PrimitiveDateTime>,
+    header_payload: HeaderPayload,
+    frm_suggestion: Option<storage_enums::FrmSuggestion>,
+    business_profile: &'a domain::Profile,
+    is_retry_payment: bool,
+    all_keys_required: Option<bool>,
+    merchant_connector_account: helpers::MerchantConnectorAccountType,
+    router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    tokenization_action: TokenizationAction,
+) -> RouterResult<(
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    helpers::MerchantConnectorAccountType,
+)>
+where
+    F: Send + Clone + Sync + 'static + 'a,
+    RouterDReq: Send + Sync + Clone + 'static + 'a + serde::Serialize,
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone + serde::Serialize,
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    router_env::logger::info!(
+        "Processing payment through Direct gateway system - payment_id={}, attempt_id={}",
+        payment_data
+            .get_payment_intent()
+            .payment_id
+            .get_string_repr(),
+        payment_data.get_payment_attempt().attempt_id
+    );
+
+    // Update feature metadata to track Direct routing usage for stickiness
+    update_gateway_system_in_feature_metadata(payment_data, GatewaySystem::Direct)?;
+
+    call_connector_service(
+        state,
+        req_state,
+        merchant_context,
+        connector,
+        operation,
+        payment_data,
+        customer,
+        call_connector_action,
+        validate_result,
+        schedule_time,
+        header_payload,
+        frm_suggestion,
+        business_profile,
+        is_retry_payment,
+        all_keys_required,
+        merchant_connector_account,
+        router_data,
+        tokenization_action,
+    )
+    .await
+}
+
+#[cfg(feature = "v1")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+// Helper function to process through Direct with Shadow UCS
+async fn process_through_direct_with_shadow_unified_connector_service<
+    'a,
+    F,
+    RouterDReq,
+    ApiRequest,
+    D,
+>(
+    state: &'a SessionState,
+    req_state: ReqState,
+    merchant_context: &'a domain::MerchantContext,
+    connector: api::ConnectorData,
+    operation: &'a BoxedOperation<'a, F, ApiRequest, D>,
+    payment_data: &'a mut D,
+    customer: &Option<domain::Customer>,
+    call_connector_action: CallConnectorAction,
+    validate_result: &'a operations::ValidateResult,
+    schedule_time: Option<time::PrimitiveDateTime>,
+    header_payload: HeaderPayload,
+    frm_suggestion: Option<storage_enums::FrmSuggestion>,
+    business_profile: &'a domain::Profile,
+    is_retry_payment: bool,
+    all_keys_required: Option<bool>,
+    merchant_connector_account: helpers::MerchantConnectorAccountType,
+    router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    tokenization_action: TokenizationAction,
+) -> RouterResult<(
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    helpers::MerchantConnectorAccountType,
+)>
+where
+    F: Send + Clone + Sync + 'static + 'a,
+    RouterDReq: Send + Sync + Clone + 'static + 'a + serde::Serialize,
+    D: OperationSessionGetters<F> + OperationSessionSetters<F> + Send + Sync + Clone,
+    D: ConstructFlowSpecificData<F, RouterDReq, router_types::PaymentsResponseData>,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone + serde::Serialize,
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    router_env::logger::info!(
+        "Processing payment through Direct gateway system with UCS in shadow mode - payment_id={}, attempt_id={}",
+        payment_data.get_payment_intent().payment_id.get_string_repr(),
+        payment_data.get_payment_attempt().attempt_id
+    );
+
+    // Clone data needed for shadow UCS call
+    let unified_connector_service_router_data = router_data.clone();
+    let unified_connector_service_merchant_connector_account = merchant_connector_account.clone();
+    let unified_connector_service_merchant_context = merchant_context.clone();
+    let unified_connector_service_header_payload = header_payload.clone();
+    let unified_connector_service_state = state.clone();
+
+    let lineage_ids = grpc_client::LineageIds::new(
+        business_profile.merchant_id.clone(),
+        business_profile.get_id().clone(),
+    );
+
+    // Update feature metadata to track Direct routing usage for stickiness
+    update_gateway_system_in_feature_metadata(payment_data, GatewaySystem::Direct)?;
+
+    // Call Direct connector service
+    let result = call_connector_service(
+        state,
+        req_state,
+        merchant_context,
+        connector,
+        operation,
+        payment_data,
+        customer,
+        call_connector_action,
+        validate_result,
+        schedule_time,
+        header_payload,
+        frm_suggestion,
+        business_profile,
+        is_retry_payment,
+        all_keys_required,
+        merchant_connector_account,
+        router_data,
+        tokenization_action,
+    )
+    .await?;
+
+    // Spawn shadow UCS call in background
+    let direct_router_data = result.0.clone();
+    tokio::spawn(async move {
+        execute_shadow_unified_connector_service_call(
+            unified_connector_service_state,
+            unified_connector_service_router_data,
+            direct_router_data,
+            unified_connector_service_header_payload,
+            lineage_ids,
+            unified_connector_service_merchant_connector_account,
+            unified_connector_service_merchant_context,
+        )
+        .await
+    });
+
+    Ok(result)
+}
+
+#[cfg(feature = "v1")]
+#[allow(clippy::too_many_arguments)]
+#[instrument(skip_all)]
+// Helper function to execute shadow UCS call
+async fn execute_shadow_unified_connector_service_call<F, RouterDReq>(
+    state: SessionState,
+    mut unified_connector_service_router_data: RouterData<
+        F,
+        RouterDReq,
+        router_types::PaymentsResponseData,
+    >,
+    direct_router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    header_payload: HeaderPayload,
+    lineage_ids: grpc_client::LineageIds,
+    merchant_connector_account: helpers::MerchantConnectorAccountType,
+    merchant_context: domain::MerchantContext,
+) where
+    F: Send + Clone + Sync + 'static,
+    RouterDReq: Send + Sync + Clone + 'static + serde::Serialize,
+    RouterData<F, RouterDReq, router_types::PaymentsResponseData>:
+        Feature<F, RouterDReq> + Send + Clone + serde::Serialize,
+    dyn api::Connector:
+        services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
+{
+    // Call UCS in shadow mode
+    let _unified_connector_service_result = unified_connector_service_router_data
+        .call_unified_connector_service(
+            &state,
+            &header_payload,
+            lineage_ids,
+            merchant_connector_account,
+            &merchant_context,
+            ExecutionMode::Shadow, // Shadow mode for UCS
+        )
+        .await
+        .map_err(|e| logger::debug!("Shadow UCS call failed: {:?}", e));
+
+    // Compare results
+    match serialize_router_data_and_send_to_comparison_service(
+        &state,
+        direct_router_data,
+        unified_connector_service_router_data,
+    )
     .await
+    {
+        Ok(_) => logger::debug!("Shadow UCS comparison completed successfully"),
+        Err(e) => logger::debug!("Shadow UCS comparison failed: {:?}", e),
+    }
+}
+
+async fn serialize_router_data_and_send_to_comparison_service<F, RouterDReq>(
+    state: &SessionState,
+    hyperswitch_router_data: RouterData<F, RouterDReq, router_types::PaymentsResponseData>,
+    unified_connector_service_router_data: RouterData<
+        F,
+        RouterDReq,
+        router_types::PaymentsResponseData,
+    >,
+) -> RouterResult<()>
+where
+    F: Send + Clone + Sync + 'static,
+    RouterDReq: Send + Sync + Clone + 'static + serde::Serialize,
+{
+    logger::info!("Simulating UCS call for shadow mode comparison");
+    let hyperswitch_data = match serde_json::to_value(hyperswitch_router_data) {
+        Ok(data) => Secret::new(data),
+        Err(_) => {
+            logger::debug!("Failed to serialize HS router data");
+            return Ok(());
+        }
+    };
+
+    let unified_connector_service_data =
+        match serde_json::to_value(unified_connector_service_router_data) {
+            Ok(data) => Secret::new(data),
+            Err(_) => {
+                logger::debug!("Failed to serialize UCS router data");
+                return Ok(());
+            }
+        };
+
+    let comparison_data = ComparisonData {
+        hyperswitch_data,
+        unified_connector_service_data,
+    };
+    let _ = send_comparison_data(state, comparison_data)
+        .await
+        .map_err(|e| {
+            logger::debug!("Failed to send comparison data: {:?}", e);
+        });
+    Ok(())
 }
 
 async fn record_time_taken_with<F, Fut, R>(f: F) -> RouterResult<R>
@@ -4818,7 +5171,10 @@ where
     )
     .await?;
 
-    let (connector_request, should_continue_further) = if !should_call_unified_connector_service {
+    let (connector_request, should_continue_further) = if matches!(
+        should_call_unified_connector_service,
+        GatewaySystem::Direct | GatewaySystem::ShadowUnifiedConnectorService
+    ) {
         let mut should_continue_further = true;
 
         let should_continue = match router_data
@@ -4875,7 +5231,7 @@ where
         .await?;
 
     record_time_taken_with(|| async {
-        if should_call_unified_connector_service {
+        if matches!(should_call_unified_connector_service, GatewaySystem::UnifiedConnectorService) {
             router_env::logger::info!(
                 "Processing payment through UCS gateway system- payment_id={}, attempt_id={}",
                 payment_data.get_payment_intent().id.get_string_repr(),
@@ -4890,6 +5246,7 @@ where
                     lineage_ids,
                     merchant_connector_account_type_details.clone(),
                     merchant_context,
+                    ExecutionMode::Primary, // UCS is called in primary mode
                 )
                 .await?;
 
@@ -4941,14 +5298,14 @@ where
         services::api::ConnectorIntegration<F, RouterDReq, router_types::PaymentsResponseData>,
 {
     record_time_taken_with(|| async {
-        if should_call_unified_connector_service(
+        let execution = should_call_unified_connector_service(
             state,
             merchant_context,
             &router_data,
             Some(payment_data),
         )
-        .await?
-        {
+        .await?;
+        if matches!(execution, GatewaySystem::UnifiedConnectorService) {
             router_env::logger::info!(
                 "Executing payment through UCS gateway system - payment_id={}, attempt_id={}",
                 payment_data.get_payment_intent().id.get_string_repr(),
@@ -4993,16 +5350,25 @@ where
                     lineage_ids,
                     merchant_connector_account_type_details.clone(),
                     merchant_context,
+                    ExecutionMode::Primary, //UCS is called in primary mode
                 )
                 .await?;
 
             Ok(router_data)
         } else {
-            router_env::logger::info!(
-                "Processing payment through Direct gateway system - payment_id={}, attempt_id={}",
-                payment_data.get_payment_intent().id.get_string_repr(),
-                payment_data.get_payment_attempt().id.get_string_repr()
-            );
+            if matches!(execution, GatewaySystem::ShadowUnifiedConnectorService) {
+                router_env::logger::info!(
+                    "Shadow UCS mode not implemented in v2, processing through direct path - payment_id={}, attempt_id={}",
+                    payment_data.get_payment_intent().id.get_string_repr(),
+                    payment_data.get_payment_attempt().id.get_string_repr()
+                );
+            } else {
+                router_env::logger::info!(
+                    "Processing payment through Direct gateway system - payment_id={}, attempt_id={}",
+                    payment_data.get_payment_intent().id.get_string_repr(),
+                    payment_data.get_payment_attempt().id.get_string_repr()
+                );
+            }
 
             call_connector_service(
                 state,
@@ -5094,6 +5460,7 @@ where
                 merchant_connector_account_type_details.clone(),
                 external_vault_merchant_connector_account_type_details.clone(),
                 merchant_context,
+                ExecutionMode::Primary, //UCS is called in primary mode
             )
             .await?;
 
diff --git a/crates/router/src/core/payments/flows.rs b/crates/router/src/core/payments/flows.rs
index 3fd25d05d6e..612f0e36176 100644
--- a/crates/router/src/core/payments/flows.rs
+++ b/crates/router/src/core/payments/flows.rs
@@ -16,6 +16,7 @@ pub mod setup_mandate_flow;
 pub mod update_metadata_flow;
 
 use async_trait::async_trait;
+use common_enums::{self, ExecutionMode};
 use common_types::payments::CustomerAcceptance;
 use external_services::grpc_client;
 #[cfg(all(feature = "v2", feature = "revenue_recovery"))]
@@ -217,6 +218,7 @@ pub trait Feature<F, T> {
         #[cfg(feature = "v2")]
         _merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         _merchant_context: &domain::MerchantContext,
+        _unified_connector_service_execution_mode: ExecutionMode,
     ) -> RouterResult<()>
     where
         F: Clone,
@@ -235,6 +237,7 @@ pub trait Feature<F, T> {
         _merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         _external_vault_merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         _merchant_context: &domain::MerchantContext,
+        _unified_connector_service_execution_mode: ExecutionMode,
     ) -> RouterResult<()>
     where
         F: Clone,
diff --git a/crates/router/src/core/payments/flows/authorize_flow.rs b/crates/router/src/core/payments/flows/authorize_flow.rs
index 6ed1613aae2..cbb5b07f857 100644
--- a/crates/router/src/core/payments/flows/authorize_flow.rs
+++ b/crates/router/src/core/payments/flows/authorize_flow.rs
@@ -533,6 +533,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
         #[cfg(feature = "v2")]
         merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         merchant_context: &domain::MerchantContext,
+        unified_connector_service_execution_mode: enums::ExecutionMode,
     ) -> RouterResult<()> {
         if self.request.mandate_id.is_some() {
             Box::pin(call_unified_connector_service_repeat_payment(
@@ -542,6 +543,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                 lineage_ids,
                 merchant_connector_account,
                 merchant_context,
+                unified_connector_service_execution_mode,
             ))
             .await
         } else {
@@ -552,6 +554,7 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                 lineage_ids,
                 merchant_connector_account,
                 merchant_context,
+                unified_connector_service_execution_mode,
             ))
             .await
         }
@@ -854,6 +857,7 @@ async fn call_unified_connector_service_authorize(
     #[cfg(feature = "v1")] merchant_connector_account: helpers::MerchantConnectorAccountType,
     #[cfg(feature = "v2")] merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
     merchant_context: &domain::MerchantContext,
+    unified_connector_service_execution_mode: enums::ExecutionMode,
 ) -> RouterResult<()> {
     let client = state
         .grpc_client
@@ -881,7 +885,7 @@ async fn call_unified_connector_service_authorize(
         .flatten()
         .map(ucs_types::UcsReferenceId::Payment);
     let headers_builder = state
-        .get_grpc_headers_ucs()
+        .get_grpc_headers_ucs(unified_connector_service_execution_mode)
         .external_vault_proxy_metadata(None)
         .merchant_reference_id(merchant_order_reference_id)
         .lineage_ids(lineage_ids);
@@ -943,6 +947,7 @@ async fn call_unified_connector_service_repeat_payment(
     #[cfg(feature = "v1")] merchant_connector_account: helpers::MerchantConnectorAccountType,
     #[cfg(feature = "v2")] merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
     merchant_context: &domain::MerchantContext,
+    unified_connector_service_execution_mode: enums::ExecutionMode,
 ) -> RouterResult<()> {
     let client = state
         .grpc_client
@@ -970,7 +975,7 @@ async fn call_unified_connector_service_repeat_payment(
         .flatten()
         .map(ucs_types::UcsReferenceId::Payment);
     let headers_builder = state
-        .get_grpc_headers_ucs()
+        .get_grpc_headers_ucs(unified_connector_service_execution_mode)
         .external_vault_proxy_metadata(None)
         .merchant_reference_id(merchant_order_reference_id)
         .lineage_ids(lineage_ids);
diff --git a/crates/router/src/core/payments/flows/external_proxy_flow.rs b/crates/router/src/core/payments/flows/external_proxy_flow.rs
index 601e3ee5c6e..4e66c5deaca 100644
--- a/crates/router/src/core/payments/flows/external_proxy_flow.rs
+++ b/crates/router/src/core/payments/flows/external_proxy_flow.rs
@@ -374,6 +374,7 @@ impl Feature<api::ExternalVaultProxy, types::ExternalVaultProxyPaymentsData>
         merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         external_vault_merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         merchant_context: &domain::MerchantContext,
+        unified_connector_service_execution_mode: enums::ExecutionMode,
     ) -> RouterResult<()> {
         let client = state
             .grpc_client
@@ -411,7 +412,7 @@ impl Feature<api::ExternalVaultProxy, types::ExternalVaultProxyPaymentsData>
             .flatten()
             .map(ucs_types::UcsReferenceId::Payment);
         let headers_builder = state
-            .get_grpc_headers_ucs()
+            .get_grpc_headers_ucs(unified_connector_service_execution_mode)
             .external_vault_proxy_metadata(Some(external_vault_proxy_metadata))
             .merchant_reference_id(merchant_order_reference_id)
             .lineage_ids(lineage_ids);
diff --git a/crates/router/src/core/payments/flows/psync_flow.rs b/crates/router/src/core/payments/flows/psync_flow.rs
index eaaa76e3b34..e54a5fd173f 100644
--- a/crates/router/src/core/payments/flows/psync_flow.rs
+++ b/crates/router/src/core/payments/flows/psync_flow.rs
@@ -1,6 +1,7 @@
 use std::{collections::HashMap, str::FromStr};
 
 use async_trait::async_trait;
+use common_enums::{self, enums};
 use common_utils::{id_type, ucs_types};
 use error_stack::ResultExt;
 use external_services::grpc_client;
@@ -234,6 +235,7 @@ impl Feature<api::PSync, types::PaymentsSyncData>
         #[cfg(feature = "v2")]
         merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         merchant_context: &domain::MerchantContext,
+        unified_connector_service_execution_mode: enums::ExecutionMode,
     ) -> RouterResult<()> {
         let connector_name = self.connector.clone();
         let connector_enum = common_enums::connector_enums::Connector::from_str(&connector_name)
@@ -285,7 +287,7 @@ impl Feature<api::PSync, types::PaymentsSyncData>
             .flatten()
             .map(ucs_types::UcsReferenceId::Payment);
         let header_payload = state
-            .get_grpc_headers_ucs()
+            .get_grpc_headers_ucs(unified_connector_service_execution_mode)
             .external_vault_proxy_metadata(None)
             .merchant_reference_id(merchant_reference_id)
             .lineage_ids(lineage_ids);
diff --git a/crates/router/src/core/payments/flows/setup_mandate_flow.rs b/crates/router/src/core/payments/flows/setup_mandate_flow.rs
index c1c243064c2..2bc69b85939 100644
--- a/crates/router/src/core/payments/flows/setup_mandate_flow.rs
+++ b/crates/router/src/core/payments/flows/setup_mandate_flow.rs
@@ -1,6 +1,7 @@
 use std::str::FromStr;
 
 use async_trait::async_trait;
+use common_enums::{self, enums};
 use common_types::payments as common_payments_types;
 use common_utils::{id_type, ucs_types};
 use error_stack::ResultExt;
@@ -274,6 +275,7 @@ impl Feature<api::SetupMandate, types::SetupMandateRequestData> for types::Setup
         #[cfg(feature = "v2")]
         merchant_connector_account: domain::MerchantConnectorAccountTypeDetails,
         merchant_context: &domain::MerchantContext,
+        unified_connector_service_execution_mode: enums::ExecutionMode,
     ) -> RouterResult<()> {
         let client = state
             .grpc_client
@@ -303,7 +305,7 @@ impl Feature<api::SetupMandate, types::SetupMandateRequestData> for types::Setup
             .flatten()
             .map(ucs_types::UcsReferenceId::Payment);
         let header_payload = state
-            .get_grpc_headers_ucs()
+            .get_grpc_headers_ucs(unified_connector_service_execution_mode)
             .external_vault_proxy_metadata(None)
             .merchant_reference_id(merchant_reference_id)
             .lineage_ids(lineage_ids);
diff --git a/crates/router/src/core/payments/retry.rs b/crates/router/src/core/payments/retry.rs
index 1e7c0ad5460..367abc4f235 100644
--- a/crates/router/src/core/payments/retry.rs
+++ b/crates/router/src/core/payments/retry.rs
@@ -32,7 +32,7 @@ use crate::{
 #[instrument(skip_all)]
 #[allow(clippy::too_many_arguments)]
 #[cfg(feature = "v1")]
-pub async fn do_gsm_actions<F, ApiRequest, FData, D>(
+pub async fn do_gsm_actions<'a, F, ApiRequest, FData, D>(
     state: &app::SessionState,
     req_state: ReqState,
     payment_data: &mut D,
@@ -48,8 +48,8 @@ pub async fn do_gsm_actions<F, ApiRequest, FData, D>(
     business_profile: &domain::Profile,
 ) -> RouterResult<types::RouterData<F, FData, types::PaymentsResponseData>>
 where
-    F: Clone + Send + Sync,
-    FData: Send + Sync + types::Capturable,
+    F: Clone + Send + Sync + 'static + 'a,
+    FData: Send + Sync + types::Capturable + Clone + 'static + 'a + serde::Serialize,
     payments::PaymentResponse: operations::Operation<F, FData>,
     D: payments::OperationSessionGetters<F>
         + payments::OperationSessionSetters<F>
@@ -338,14 +338,14 @@ fn get_flow_name<F>() -> RouterResult<String> {
 #[cfg(feature = "v1")]
 #[allow(clippy::too_many_arguments)]
 #[instrument(skip_all)]
-pub async fn do_retry<F, ApiRequest, FData, D>(
-    state: &routes::SessionState,
+pub async fn do_retry<'a, F, ApiRequest, FData, D>(
+    state: &'a routes::SessionState,
     req_state: ReqState,
-    connector: &api::ConnectorData,
-    operation: &operations::BoxedOperation<'_, F, ApiRequest, D>,
-    customer: &Option<domain::Customer>,
+    connector: &'a api::ConnectorData,
+    operation: &'a operations::BoxedOperation<'a, F, ApiRequest, D>,
+    customer: &'a Option<domain::Customer>,
     merchant_context: &domain::MerchantContext,
-    payment_data: &mut D,
+    payment_data: &'a mut D,
     router_data: types::RouterData<F, FData, types::PaymentsResponseData>,
     validate_result: &operations::ValidateResult,
     schedule_time: Option<time::PrimitiveDateTime>,
@@ -356,8 +356,8 @@ pub async fn do_retry<F, ApiRequest, FData, D>(
     routing_decision: Option<routing_helpers::RoutingDecisionData>,
 ) -> RouterResult<types::RouterData<F, FData, types::PaymentsResponseData>>
 where
-    F: Clone + Send + Sync,
-    FData: Send + Sync + types::Capturable,
+    F: Clone + Send + Sync + 'static + 'a,
+    FData: Send + Sync + types::Capturable + Clone + 'static + 'a + serde::Serialize,
     payments::PaymentResponse: operations::Operation<F, FData>,
     D: payments::OperationSessionGetters<F>
         + payments::OperationSessionSetters<F>
diff --git a/crates/router/src/core/unified_connector_service.rs b/crates/router/src/core/unified_connector_service.rs
index fbfe9cec635..04a3a2e4ef2 100644
--- a/crates/router/src/core/unified_connector_service.rs
+++ b/crates/router/src/core/unified_connector_service.rs
@@ -3,15 +3,25 @@ use std::{str::FromStr, time::Instant};
 use api_models::admin;
 #[cfg(feature = "v2")]
 use base64::Engine;
-use common_enums::{connector_enums::Connector, AttemptStatus, GatewaySystem, PaymentMethodType};
+use common_enums::{
+    connector_enums::Connector, AttemptStatus, ExecutionMode, GatewaySystem, PaymentMethodType,
+};
 #[cfg(feature = "v2")]
 use common_utils::consts::BASE64_ENGINE;
-use common_utils::{errors::CustomResult, ext_traits::ValueExt};
+use common_utils::{
+    consts::X_FLOW_NAME,
+    errors::CustomResult,
+    ext_traits::ValueExt,
+    request::{Method, RequestBuilder, RequestContent},
+};
 use diesel_models::types::FeatureMetadata;
 use error_stack::ResultExt;
-use external_services::grpc_client::{
-    unified_connector_service::{ConnectorAuthMetadata, UnifiedConnectorServiceError},
-    LineageIds,
+use external_services::{
+    grpc_client::{
+        unified_connector_service::{ConnectorAuthMetadata, UnifiedConnectorServiceError},
+        LineageIds,
+    },
+    http_client,
 };
 use hyperswitch_connectors::utils::CardData;
 #[cfg(feature = "v2")]
@@ -46,6 +56,7 @@ use crate::{
         utils::get_flow_name,
     },
     events::connector_api_logs::ConnectorEvent,
+    headers::{CONTENT_TYPE, X_REQUEST_ID},
     routes::SessionState,
     types::transformers::ForeignTryFrom,
 };
@@ -64,90 +75,87 @@ type UnifiedConnectorServiceResult = CustomResult<
     UnifiedConnectorServiceError,
 >;
 
-/// Generic version of should_call_unified_connector_service that works with any type
-/// implementing OperationSessionGetters trait
 pub async fn should_call_unified_connector_service<F: Clone, T, D>(
     state: &SessionState,
     merchant_context: &MerchantContext,
     router_data: &RouterData<F, T, PaymentsResponseData>,
     payment_data: Option<&D>,
-) -> RouterResult<bool>
+) -> RouterResult<GatewaySystem>
 where
     D: OperationSessionGetters<F>,
 {
-    // Check basic UCS availability first
-    if state.grpc_client.unified_connector_service_client.is_none() {
+    // Extract context information
+    let merchant_id = merchant_context
+        .get_merchant_account()
+        .get_id()
+        .get_string_repr();
+
+    let connector_name = &router_data.connector;
+    let connector_enum = Connector::from_str(connector_name)
+        .change_context(errors::ApiErrorResponse::IncorrectConnectorNameGiven)
+        .attach_printable_lazy(|| format!("Failed to parse connector name: {}", connector_name))?;
+
+    let payment_method = router_data.payment_method.to_string();
+    let flow_name = get_flow_name::<F>()?;
+
+    // Compute all relevant conditions
+    let ucs_client_available = state.grpc_client.unified_connector_service_client.is_some();
+    let ucs_enabled = is_ucs_enabled(state, consts::UCS_ENABLED).await;
+
+    // Log if UCS client is not available
+    if !ucs_client_available {
         router_env::logger::debug!(
-            "Unified Connector Service client is not available, skipping UCS decision"
+            "UCS client not available - merchant_id={}, connector={}",
+            merchant_id,
+            connector_name
         );
-        return Ok(false);
     }
 
-    let ucs_config_key = consts::UCS_ENABLED;
-    if !is_ucs_enabled(state, ucs_config_key).await {
+    // Log if UCS is not enabled
+    if !ucs_enabled {
         router_env::logger::debug!(
-            "Unified Connector Service is not enabled, skipping UCS decision"
+            "UCS not enabled in configuration - merchant_id={}, connector={}",
+            merchant_id,
+            connector_name
         );
-        return Ok(false);
     }
 
-    // Apply stickiness logic if payment_data is available
-    if let Some(payment_data) = payment_data {
-        let previous_gateway_system = extract_gateway_system_from_payment_intent(payment_data);
+    let ucs_config = state.conf.grpc_client.unified_connector_service.as_ref();
 
-        match previous_gateway_system {
-            Some(GatewaySystem::UnifiedConnectorService) => {
-                // Payment intent previously used UCS, maintain stickiness to UCS
-                router_env::logger::info!(
-                    "Payment gateway system decision: UCS (sticky) - payment intent previously used UCS"
-                );
-                return Ok(true);
-            }
-            Some(GatewaySystem::Direct) => {
-                // Payment intent previously used Direct, maintain stickiness to Direct (return false for UCS)
-                router_env::logger::info!(
-                    "Payment gateway system decision: Direct (sticky) - payment intent previously used Direct"
-                );
-                return Ok(false);
-            }
-            None => {
-                // No previous gateway system set, continue with normal gateway system logic
-                router_env::logger::debug!(
-                    "UCS stickiness: No previous gateway system set, applying normal gateway system logic"
-                );
-            }
-        }
+    // Log if UCS configuration is missing
+    if ucs_config.is_none() {
+        router_env::logger::debug!(
+            "UCS configuration not found - merchant_id={}, connector={}",
+            merchant_id,
+            connector_name
+        );
     }
 
-    // Continue with normal UCS gateway system logic
-    let merchant_id = merchant_context
-        .get_merchant_account()
-        .get_id()
-        .get_string_repr();
+    let ucs_only_connector =
+        ucs_config.is_some_and(|config| config.ucs_only_connectors.contains(&connector_enum));
 
-    let connector_name = router_data.connector.clone();
-    let connector_enum = Connector::from_str(&connector_name)
-        .change_context(errors::ApiErrorResponse::IncorrectConnectorNameGiven)?;
+    let previous_gateway = payment_data.and_then(extract_gateway_system_from_payment_intent);
 
-    let payment_method = router_data.payment_method.to_string();
-    let flow_name = get_flow_name::<F>()?;
-
-    let is_ucs_only_connector = state
-        .conf
-        .grpc_client
-        .unified_connector_service
-        .as_ref()
-        .is_some_and(|config| config.ucs_only_connectors.contains(&connector_enum));
-
-    if is_ucs_only_connector {
-        router_env::logger::info!(
-            "Payment gateway system decision: UCS (forced) - merchant_id={}, connector={}, payment_method={}, flow={}",
-            merchant_id, connector_name, payment_method, flow_name
-        );
-        return Ok(true);
+    // Log previous gateway state
+    match previous_gateway {
+        Some(gateway) => {
+            router_env::logger::debug!(
+                "Previous gateway system found: {:?} - merchant_id={}, connector={}",
+                gateway,
+                merchant_id,
+                connector_name
+            );
+        }
+        None => {
+            router_env::logger::debug!(
+                "No previous gateway system found (new payment) - merchant_id={}, connector={}",
+                merchant_id,
+                connector_name
+            );
+        }
     }
 
-    let config_key = format!(
+    let rollout_key = format!(
         "{}_{}_{}_{}_{}",
         consts::UCS_ROLLOUT_PERCENT_CONFIG_PREFIX,
         merchant_id,
@@ -155,23 +163,179 @@ where
         payment_method,
         flow_name
     );
+    let shadow_rollout_key = format!("{}_shadow", rollout_key);
 
-    let should_execute = should_execute_based_on_rollout(state, &config_key).await?;
+    let rollout_enabled = should_execute_based_on_rollout(state, &rollout_key).await?;
+    let shadow_rollout_enabled =
+        should_execute_based_on_rollout(state, &shadow_rollout_key).await?;
 
-    // Log gateway system decision
-    if should_execute {
-        router_env::logger::info!(
-            "Payment gateway system decision: UCS - merchant_id={}, connector={}, payment_method={}, flow={}",
-            merchant_id, connector_name, payment_method, flow_name
-        );
-    } else {
-        router_env::logger::info!(
-            "Payment gateway system decision: Direct - merchant_id={}, connector={}, payment_method={}, flow={}",
-            merchant_id, connector_name, payment_method, flow_name
-        );
-    }
+    router_env::logger::debug!(
+        "Rollout status - rollout_enabled={}, shadow_rollout_enabled={}, rollout_key={}, merchant_id={}, connector={}",
+        rollout_enabled,
+        shadow_rollout_enabled,
+        rollout_key,
+        merchant_id,
+        connector_name
+    );
 
-    Ok(should_execute)
+    // Single decision point using pattern matching
+    // Tuple structure: (ucs_infrastructure_ready, ucs_only_connector, previous_gateway, shadow_rollout_enabled, rollout_enabled)
+    let decision = match (
+        ucs_client_available && ucs_enabled,
+        ucs_only_connector,
+        previous_gateway,
+        shadow_rollout_enabled,
+        rollout_enabled,
+    ) {
+        // ==================== DIRECT GATEWAY DECISIONS ====================
+        // All patterns that result in Direct routing
+
+        // UCS infrastructure not available (any configuration)
+        // - Client not available OR not enabled
+        // - Regardless of: ucs_only_connector, previous_gateway, rollouts
+        (false, _, _, _, _) |
+
+        // UCS available but no rollouts active and no previous gateway
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - New payment (no previous gateway)
+        // - No shadow rollout
+        // - No full rollout
+        (true, false, None, false, false) |
+
+        // UCS available, continuing with Direct, no rollouts
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Direct
+        // - No shadow rollout
+        // - No full rollout
+        (true, false, Some(GatewaySystem::Direct), false, false) |
+
+        // UCS available, previous Shadow but rollout ended
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Shadow
+        // - No shadow rollout (ended)
+        // - No full rollout
+        (true, false, Some(GatewaySystem::ShadowUnifiedConnectorService), false, false) => {
+            router_env::logger::debug!(
+                "Routing to Direct: ucs_ready={}, ucs_only={}, previous={:?}, shadow_rollout={}, full_rollout={} - merchant_id={}, connector={}",
+                ucs_client_available && ucs_enabled,
+                ucs_only_connector,
+                previous_gateway,
+                shadow_rollout_enabled,
+                rollout_enabled,
+                merchant_id,
+                connector_name
+            );
+            GatewaySystem::Direct
+        }
+
+        // ==================== SHADOW UCS DECISIONS ====================
+        // All patterns that result in Shadow UCS routing
+
+        // Shadow rollout for new payment (no previous gateway)
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - No previous gateway
+        // - Shadow rollout enabled
+        // - Full rollout: any (false or true, shadow takes precedence)
+        (true, false, None, true, _) |
+
+        // Shadow rollout with previous Direct gateway
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Direct
+        // - Shadow rollout enabled
+        // - Full rollout: any (shadow takes precedence)
+        (true, false, Some(GatewaySystem::Direct), true, _) |
+
+        // Shadow rollout with previous Shadow gateway (continuation)
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Shadow
+        // - Shadow rollout enabled
+        // - Full rollout: any
+        (true, false, Some(GatewaySystem::ShadowUnifiedConnectorService), true, _) => {
+            router_env::logger::debug!(
+                "Routing to ShadowUnifiedConnectorService: ucs_ready={}, ucs_only={}, previous={:?}, shadow_rollout={}, full_rollout={} - merchant_id={}, connector={}",
+                ucs_client_available && ucs_enabled,
+                ucs_only_connector,
+                previous_gateway,
+                shadow_rollout_enabled,
+                rollout_enabled,
+                merchant_id,
+                connector_name
+            );
+            GatewaySystem::ShadowUnifiedConnectorService
+        }
+
+        // ==================== UNIFIED CONNECTOR SERVICE DECISIONS ====================
+        // All patterns that result in UCS routing
+
+        // UCS-only connector (mandatory UCS)
+        // - Infrastructure ready
+        // - UCS-only connector flag set
+        // - Any previous gateway
+        // - Any shadow rollout state
+        // - Any full rollout state
+        (true, true, _, _, _) |
+
+        // Sticky routing: Continue with previous UCS
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was UCS
+        // - Any shadow rollout state (doesn't affect existing UCS)
+        // - Any full rollout state
+        (true, false, Some(GatewaySystem::UnifiedConnectorService), _, _) |
+
+        // Full rollout: New payment
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - No previous gateway
+        // - No shadow rollout (shadow would take precedence)
+        // - Full rollout enabled
+        (true, false, None, false, true) |
+
+        // Full rollout: Switch from Direct
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Direct
+        // - No shadow rollout (shadow would take precedence)
+        // - Full rollout enabled
+        (true, false, Some(GatewaySystem::Direct), false, true) |
+
+        // Full rollout: Promote from Shadow
+        // - Infrastructure ready
+        // - Not a UCS-only connector
+        // - Previous gateway was Shadow
+        // - Shadow rollout ended (now false)
+        // - Full rollout enabled
+        (true, false, Some(GatewaySystem::ShadowUnifiedConnectorService), false, true) => {
+            router_env::logger::debug!(
+                "Routing to UnifiedConnectorService: ucs_ready={}, ucs_only={}, previous={:?}, shadow_rollout={}, full_rollout={} - merchant_id={}, connector={}",
+                ucs_client_available && ucs_enabled,
+                ucs_only_connector,
+                previous_gateway,
+                shadow_rollout_enabled,
+                rollout_enabled,
+                merchant_id,
+                connector_name
+            );
+            GatewaySystem::UnifiedConnectorService
+        }
+    };
+
+    router_env::logger::info!(
+        "Payment gateway system decision: {:?} - merchant_id={}, connector={}, payment_method={}, flow={}",
+        decision,
+        merchant_id,
+        connector_name,
+        payment_method,
+        flow_name
+    );
+
+    Ok(decision)
 }
 
 /// Extracts the gateway system from the payment intent's feature metadata
@@ -746,7 +910,7 @@ pub async fn call_unified_connector_service_for_webhook(
         .unwrap_or(consts::PROFILE_ID_UNAVAILABLE.clone());
     // Build gRPC headers
     let grpc_headers = state
-        .get_grpc_headers_ucs()
+        .get_grpc_headers_ucs(ExecutionMode::Primary)
         .lineage_ids(LineageIds::new(
             merchant_context.get_merchant_account().get_id().clone(),
             profile_id,
@@ -882,7 +1046,7 @@ where
         std::any::type_name::<T>(),
         grpc_request_body,
         "grpc://unified-connector-service".to_string(),
-        common_utils::request::Method::Post,
+        Method::Post,
         payment_id,
         merchant_id,
         state.request_id.as_ref(),
@@ -909,3 +1073,45 @@ where
 
     router_result
 }
+
+#[derive(serde::Serialize)]
+pub struct ComparisonData {
+    pub hyperswitch_data: Secret<serde_json::Value>,
+    pub unified_connector_service_data: Secret<serde_json::Value>,
+}
+
+/// Sends router data comparison to external service
+pub async fn send_comparison_data(
+    state: &SessionState,
+    comparison_data: ComparisonData,
+) -> RouterResult<()> {
+    // Check if comparison service is enabled
+    let comparison_config = match state.conf.comparison_service.as_ref() {
+        Some(comparison_config) => comparison_config,
+        None => {
+            tracing::warn!(
+                "Comparison service configuration missing, skipping comparison data send"
+            );
+            return Ok(());
+        }
+    };
+
+    let mut request = RequestBuilder::new()
+        .method(Method::Post)
+        .url(comparison_config.url.get_string_repr())
+        .header(CONTENT_TYPE, "application/json")
+        .header(X_FLOW_NAME, "router-data")
+        .set_body(RequestContent::Json(Box::new(comparison_data)))
+        .build();
+    if let Some(req_id) = &state.request_id {
+        request.add_header(X_REQUEST_ID, masking::Maskable::Normal(req_id.to_string()));
+    }
+
+    let _ = http_client::send_request(&state.conf.proxy, request, comparison_config.timeout_secs)
+        .await
+        .map_err(|e| {
+            tracing::debug!("Error sending comparison data: {:?}", e);
+        });
+
+    Ok(())
+}
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 3a46c1a9767..43e1deb4431 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -5,7 +5,7 @@ use actix_web::{web, Scope};
 use api_models::routing::RoutingRetrieveQuery;
 use api_models::routing::RuleMigrationQuery;
 #[cfg(feature = "olap")]
-use common_enums::TransactionType;
+use common_enums::{ExecutionMode, TransactionType};
 #[cfg(feature = "partial-auth")]
 use common_utils::crypto::Blake3;
 use common_utils::id_type;
@@ -157,9 +157,20 @@ impl SessionState {
             request_id: self.request_id.map(|req_id| (*req_id).to_string()),
         }
     }
-    pub fn get_grpc_headers_ucs(&self) -> GrpcHeadersUcsBuilderInitial {
+    pub fn get_grpc_headers_ucs(
+        &self,
+        unified_connector_service_execution_mode: ExecutionMode,
+    ) -> GrpcHeadersUcsBuilderInitial {
         let tenant_id = self.tenant.tenant_id.get_string_repr().to_string();
-        GrpcHeadersUcs::builder().tenant_id(tenant_id)
+        let request_id = self.request_id.map(|req_id| (*req_id).to_string());
+        let shadow_mode = match unified_connector_service_execution_mode {
+            ExecutionMode::Primary => false,
+            ExecutionMode::Shadow => true,
+        };
+        GrpcHeadersUcs::builder()
+            .tenant_id(tenant_id)
+            .request_id(request_id)
+            .shadow_mode(Some(shadow_mode))
     }
     #[cfg(all(feature = "revenue_recovery", feature = "v2"))]
     pub fn get_recovery_grpc_headers(&self) -> GrpcRecoveryHeaders {
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index 17150020186..230250400be 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -20,7 +20,9 @@ pub use client::{ApiClient, MockApiClient, ProxyClient};
 pub use common_enums::enums::PaymentAction;
 pub use common_utils::request::{ContentType, Method, Request, RequestBuilder};
 use common_utils::{
-    consts::{DEFAULT_TENANT, TENANT_HEADER, X_HS_LATENCY},
+    consts::{
+        DEFAULT_TENANT, TENANT_HEADER, X_CONNECTOR_NAME, X_FLOW_NAME, X_HS_LATENCY, X_REQUEST_ID,
+    },
     errors::{ErrorSwitch, ReportSwitchExt},
     request::RequestContent,
 };
@@ -61,7 +63,7 @@ use crate::{
     core::{
         api_locking,
         errors::{self, CustomResult},
-        payments, unified_connector_service,
+        payments, unified_connector_service, utils as core_utils,
     },
     events::{
         api_logs::{ApiEvent, ApiEventMetric, ApiEventsType},
@@ -296,10 +298,8 @@ where
                     ("connector", req.connector.to_string()),
                     (
                         "flow",
-                        std::any::type_name::<T>()
-                            .split("::")
-                            .last()
-                            .unwrap_or_default()
+                        core_utils::get_flow_name::<T>()
+                            .unwrap_or_else(|_| "UnknownFlow".to_string())
                     ),
                 ),
             );
@@ -326,7 +326,7 @@ where
             };
 
             match connector_request {
-                Some(request) => {
+                Some(mut request) => {
                     let masked_request_body = match &request.body {
                         Some(request) => match request {
                             RequestContent::Json(i)
@@ -341,6 +341,27 @@ where
                         },
                         None => serde_json::Value::Null,
                     };
+                    let flow_name = core_utils::get_flow_name::<T>()
+                        .unwrap_or_else(|_| "UnknownFlow".to_string());
+
+                    request.headers.insert((
+                        X_FLOW_NAME.to_string(),
+                        Maskable::Masked(Secret::new(flow_name.to_string())),
+                    ));
+
+                    let connector_name = req.connector.clone();
+                    request.headers.insert((
+                        X_CONNECTOR_NAME.to_string(),
+                        Maskable::Masked(Secret::new(connector_name.clone().to_string())),
+                    ));
+                    state.request_id.as_ref().map(|id| {
+                        let request_id = id.to_string();
+                        request.headers.insert((
+                            X_REQUEST_ID.to_string(),
+                            Maskable::Normal(request_id.clone()),
+                        ));
+                        request_id
+                    });
                     let request_url = request.url.clone();
                     let request_method = request.method;
                     let current_time = Instant::now();