Issue:
Bug: refactor(euclid): refactor logs for evaluation of equality for dynamic routing evaluate response



Improve the logs for better diffs between legacy and decision_engine's euclid.

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [ ] Enhancement
- [x] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description

This PR refactors the routing logic to improve how **logs are emitted and connector evaluation is handled** when dynamic routing is performed using the Decision Engine (Euclid). Key updates include:

- Replaces raw response propagation from `perform_decision_euclid_routing` with a wrapped `RoutingEvaluateResponse` containing both `output` and `evaluated_output`.
- Introduces a new function `extract_de_output_connectors()` to deserialize and normalize `output` from Euclid into a list of `ConnectorInfo`.
- Adds `transform_de_output_for_router()` to combine `output` and `evaluated_output`, ensuring deduplication and priority order.
- Enhances logging in `compare_and_log_result` to include diff details between `de_result` and `router_result` more clearly.
- Adds `TryFrom<ConnectorInfo> for DeRoutableConnectorChoice` to safely convert API inputs to internal representations.

## Diff Hunk Explanation

### `crates/router/src/core/payments/routing.rs`
- Renamed `de_euclid_connectors` → `de_evaluated_connector` to reflect the structured response.
- Wrapped `perform_decision_euclid_routing` call with success match block to:
  - Extract connectors from `output` using `extract_de_output_connectors`.
  - Merge `output` and `evaluated_output` using `transform_de_output_for_router`.
- Updated logs to emit clearer error/debug statements during extraction, transformation, and comparison.

### `crates/router/src/core/payments/routing/utils.rs`
- Refactored `perform_decision_euclid_routing` to return `RoutingEvaluateResponse` instead of just connector list.
- Added `extract_de_output_connectors`: deserializes various output types (`single`, `priority`, `volume_split`, etc.).
- Added `transform_de_output_for_router`: deduplicates connectors and maintains priority between `evaluated_output` and fallback `output`.
- Improved `compare_and_log_result` to log equality and diff details between Decision Engine result and final connector result.
- Introduced `TryFrom<ConnectorInfo>` for `DeRoutableConnectorChoice`.

This refactor sets the stage for cleaner observability and deterministic routing behavior between evaluated and fallback paths.



## 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)?
-->
The testing guidelines are present in this [doc](https://docs.google.com/document/d/1Atb-5bIgA-H_H2lS2O-plhtGUqHK4fcQ1XDXWjrcTHE/edit?usp=sharing).


## 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:

The testing guidelines are present in this [doc](https://docs.google.com/document/d/1Atb-5bIgA-H_H2lS2O-plhtGUqHK4fcQ1XDXWjrcTHE/edit?usp=sharing).

Patch:
diff --git a/crates/router/src/core/payments/routing.rs b/crates/router/src/core/payments/routing.rs
index 547f49ce552..19ddeb7f15f 100644
--- a/crates/router/src/core/payments/routing.rs
+++ b/crates/router/src/core/payments/routing.rs
@@ -48,7 +48,6 @@ use rand::SeedableRng;
 use router_env::{instrument, tracing};
 use rustc_hash::FxHashMap;
 use storage_impl::redis::cache::{CacheKey, CGRAPH_CACHE, ROUTING_CACHE};
-use utils::perform_decision_euclid_routing;
 
 #[cfg(feature = "v2")]
 use crate::core::admin;
@@ -492,33 +491,24 @@ pub async fn perform_static_routing_v1(
             .to_string(),
     };
 
-    let de_euclid_connectors = if state.conf.open_router.static_routing_enabled {
-        let routing_events_wrapper = utils::RoutingEventsWrapper::new(
-            state.tenant.tenant_id.clone(),
-            state.request_id,
-            payment_id,
-            business_profile.get_id().to_owned(),
-            business_profile.merchant_id.to_owned(),
-            "DecisionEngine: Euclid Static Routing".to_string(),
-            None,
-            true,
-            false,
-        );
-
-        perform_decision_euclid_routing(
+    // Decision of de-routing is stored
+    let de_evaluated_connector = if !state.conf.open_router.static_routing_enabled {
+        logger::debug!("decision_engine_euclid: decision_engine routing not enabled");
+        Vec::default()
+    } else {
+        utils::decision_engine_routing(
             state,
             backend_input.clone(),
-            business_profile.get_id().get_string_repr().to_string(),
-            routing_events_wrapper,
+            business_profile,
+            payment_id,
             get_merchant_fallback_config().await?,
         )
         .await
         .map_err(|e|
             // errors are ignored as this is just for diff checking as of now (optional flow).
             logger::error!(decision_engine_euclid_evaluate_error=?e, "decision_engine_euclid: error in evaluation of rule")
-        ).unwrap_or_default()
-    } else {
-        Vec::default()
+        )
+        .unwrap_or_default()
     };
 
     let (routable_connectors, routing_approach) = match cached_algorithm.as_ref() {
@@ -538,8 +528,14 @@ pub async fn perform_static_routing_v1(
         ),
     };
 
+    // Results are logged for diff(between legacy and decision_engine's euclid) and have parameters as:
+    // is_equal: verifies all output are matching in order,
+    // is_equal_length: matches length of both outputs (useful for verifying volume based routing
+    // results)
+    // de_response: response from the decision_engine's euclid
+    // hs_response: response from legacy_euclid
     utils::compare_and_log_result(
-        de_euclid_connectors.clone(),
+        de_evaluated_connector.clone(),
         routable_connectors.clone(),
         "evaluate_routing".to_string(),
     );
@@ -549,7 +545,7 @@ pub async fn perform_static_routing_v1(
             state,
             business_profile,
             routable_connectors,
-            de_euclid_connectors,
+            de_evaluated_connector,
         )
         .await,
         routing_approach,
diff --git a/crates/router/src/core/payments/routing/utils.rs b/crates/router/src/core/payments/routing/utils.rs
index d9d5ed1de26..3c7a081d41f 100644
--- a/crates/router/src/core/payments/routing/utils.rs
+++ b/crates/router/src/core/payments/routing/utils.rs
@@ -1,4 +1,7 @@
-use std::collections::{HashMap, HashSet};
+use std::{
+    collections::{HashMap, HashSet},
+    str::FromStr,
+};
 
 use api_models::{
     open_router as or_types,
@@ -9,7 +12,7 @@ use api_models::{
     },
 };
 use async_trait::async_trait;
-use common_enums::TransactionType;
+use common_enums::{RoutableConnectors, TransactionType};
 use common_utils::{
     ext_traits::{BytesExt, StringExt},
     id_type,
@@ -30,6 +33,7 @@ use serde::{Deserialize, Serialize};
 use super::RoutingResult;
 use crate::{
     core::errors,
+    db::domain,
     routes::{app::SessionStateInfo, SessionState},
     services::{self, logger},
     types::transformers::ForeignInto,
@@ -301,7 +305,7 @@ pub async fn perform_decision_euclid_routing(
     created_by: String,
     events_wrapper: RoutingEventsWrapper<RoutingEvaluateRequest>,
     fallback_output: Vec<RoutableConnectorChoice>,
-) -> RoutingResult<Vec<RoutableConnectorChoice>> {
+) -> RoutingResult<RoutingEvaluateResponse> {
     logger::debug!("decision_engine_euclid: evaluate api call for euclid routing evaluation");
 
     let mut events_wrapper = events_wrapper;
@@ -348,8 +352,254 @@ pub async fn perform_decision_euclid_routing(
 
     logger::debug!(decision_engine_euclid_response=?euclid_response,"decision_engine_euclid");
     logger::debug!(decision_engine_euclid_selected_connector=?euclid_response.evaluated_output,"decision_engine_euclid");
+    Ok(euclid_response)
+}
+
+/// This function transforms the decision_engine response in a way that's usable for further flows:
+/// It places evaluated_output connectors first, followed by remaining output connectors (no duplicates).
+pub fn transform_de_output_for_router(
+    de_output: Vec<ConnectorInfo>,
+    de_evaluated_output: Vec<RoutableConnectorChoice>,
+) -> RoutingResult<Vec<RoutableConnectorChoice>> {
+    let mut seen = HashSet::new();
+
+    // evaluated connectors on top, to ensure the fallback is based on other connectors.
+    let mut ordered = Vec::with_capacity(de_output.len() + de_evaluated_output.len());
+    for eval_conn in de_evaluated_output {
+        if seen.insert(eval_conn.connector) {
+            ordered.push(eval_conn);
+        }
+    }
+
+    // Add remaining connectors from de_output (only if not already seen), for fallback
+    for conn in de_output {
+        let key = RoutableConnectors::from_str(&conn.gateway_name).map_err(|_| {
+            errors::RoutingError::GenericConversionError {
+                from: "String".to_string(),
+                to: "RoutableConnectors".to_string(),
+            }
+        })?;
+        if seen.insert(key) {
+            let de_choice = DeRoutableConnectorChoice::try_from(conn)?;
+            ordered.push(RoutableConnectorChoice::from(de_choice));
+        }
+    }
+    Ok(ordered)
+}
+
+pub async fn decision_engine_routing(
+    state: &SessionState,
+    backend_input: BackendInput,
+    business_profile: &domain::Profile,
+    payment_id: String,
+    merchant_fallback_config: Vec<RoutableConnectorChoice>,
+) -> RoutingResult<Vec<RoutableConnectorChoice>> {
+    let routing_events_wrapper = RoutingEventsWrapper::new(
+        state.tenant.tenant_id.clone(),
+        state.request_id,
+        payment_id,
+        business_profile.get_id().to_owned(),
+        business_profile.merchant_id.to_owned(),
+        "DecisionEngine: Euclid Static Routing".to_string(),
+        None,
+        true,
+        false,
+    );
+
+    let de_euclid_evaluate_response = perform_decision_euclid_routing(
+        state,
+        backend_input.clone(),
+        business_profile.get_id().get_string_repr().to_string(),
+        routing_events_wrapper,
+        merchant_fallback_config,
+    )
+    .await;
+
+    let Ok(de_euclid_response) = de_euclid_evaluate_response else {
+        logger::error!("decision_engine_euclid_evaluation_error: error in evaluation of rule");
+        return Ok(Vec::default());
+    };
+
+    let de_output_conenctor = extract_de_output_connectors(de_euclid_response.output)
+            .map_err(|e| {
+                logger::error!(error=?e, "decision_engine_euclid_evaluation_error: Failed to extract connector from Output");
+                e
+            })?;
+
+    transform_de_output_for_router(
+            de_output_conenctor.clone(),
+            de_euclid_response.evaluated_output.clone(),
+        )
+        .map_err(|e| {
+            logger::error!(error=?e, "decision_engine_euclid_evaluation_error: failed to transform connector from de-output");
+            e
+        })
+}
+
+/// Custom deserializer for output from decision_engine, this is required as untagged enum is
+/// stored but the enum requires tagged deserialization, hence deserializing it into specific
+/// variants
+pub fn extract_de_output_connectors(
+    output_value: serde_json::Value,
+) -> RoutingResult<Vec<ConnectorInfo>> {
+    const SINGLE: &str = "straight_through";
+    const PRIORITY: &str = "priority";
+    const VOLUME_SPLIT: &str = "volume_split";
+    const VOLUME_SPLIT_PRIORITY: &str = "volume_split_priority";
+
+    let obj = output_value.as_object().ok_or_else(|| {
+        logger::error!("decision_engine_euclid_error: output is not a JSON object");
+        errors::RoutingError::OpenRouterError("Expected output to be a JSON object".into())
+    })?;
+
+    let type_str = obj.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
+        logger::error!("decision_engine_euclid_error: missing or invalid 'type' in output");
+        errors::RoutingError::OpenRouterError("Missing or invalid 'type' field in output".into())
+    })?;
+
+    match type_str {
+        SINGLE => {
+            let connector_value = obj.get("connector").ok_or_else(|| {
+                logger::error!(
+                    "decision_engine_euclid_error: missing 'connector' field for type=single"
+                );
+                errors::RoutingError::OpenRouterError(
+                    "Missing 'connector' field for single output".into(),
+                )
+            })?;
+            let connector: ConnectorInfo = serde_json::from_value(connector_value.clone())
+                .map_err(|e| {
+                    logger::error!(
+                        ?e,
+                        "decision_engine_euclid_error: Failed to parse single connector"
+                    );
+                    errors::RoutingError::OpenRouterError(
+                        "Failed to deserialize single connector".into(),
+                    )
+                })?;
+            Ok(vec![connector])
+        }
+
+        PRIORITY => {
+            let connectors_value = obj.get("connectors").ok_or_else(|| {
+                logger::error!(
+                    "decision_engine_euclid_error: missing 'connectors' field for type=priority"
+                );
+                errors::RoutingError::OpenRouterError(
+                    "Missing 'connectors' field for priority output".into(),
+                )
+            })?;
+            let connectors: Vec<ConnectorInfo> = serde_json::from_value(connectors_value.clone())
+                .map_err(|e| {
+                logger::error!(
+                    ?e,
+                    "decision_engine_euclid_error: Failed to parse connectors for priority"
+                );
+                errors::RoutingError::OpenRouterError(
+                    "Failed to deserialize priority connectors".into(),
+                )
+            })?;
+            Ok(connectors)
+        }
+
+        VOLUME_SPLIT => {
+            let splits_value = obj.get("splits").ok_or_else(|| {
+                logger::error!(
+                    "decision_engine_euclid_error: missing 'splits' field for type=volume_split"
+                );
+                errors::RoutingError::OpenRouterError(
+                    "Missing 'splits' field for volume_split output".into(),
+                )
+            })?;
 
-    Ok(euclid_response.evaluated_output)
+            // Transform each {connector, split} into {output, split}
+            let fixed_splits: Vec<_> = splits_value
+                .as_array()
+                .ok_or_else(|| {
+                    logger::error!("decision_engine_euclid_error: 'splits' is not an array");
+                    errors::RoutingError::OpenRouterError("'splits' field must be an array".into())
+                })?
+                .iter()
+                .map(|entry| {
+                    let mut entry_map = entry.as_object().cloned().ok_or_else(|| {
+                        logger::error!(
+                            "decision_engine_euclid_error: invalid split entry in volume_split"
+                        );
+                        errors::RoutingError::OpenRouterError(
+                            "Invalid entry in splits array".into(),
+                        )
+                    })?;
+                    if let Some(connector) = entry_map.remove("connector") {
+                        entry_map.insert("output".to_string(), connector);
+                    }
+                    Ok::<_, error_stack::Report<errors::RoutingError>>(serde_json::Value::Object(
+                        entry_map,
+                    ))
+                })
+                .collect::<Result<Vec<_>, _>>()?;
+
+            let splits: Vec<VolumeSplit<ConnectorInfo>> =
+                serde_json::from_value(serde_json::Value::Array(fixed_splits)).map_err(|e| {
+                    logger::error!(
+                        ?e,
+                        "decision_engine_euclid_error: Failed to parse volume_split"
+                    );
+                    errors::RoutingError::OpenRouterError(
+                        "Failed to deserialize volume_split connectors".into(),
+                    )
+                })?;
+
+            Ok(splits.into_iter().map(|s| s.output).collect())
+        }
+
+        VOLUME_SPLIT_PRIORITY => {
+            let splits_value = obj.get("splits").ok_or_else(|| {
+                logger::error!("decision_engine_euclid_error: missing 'splits' field for type=volume_split_priority");
+                errors::RoutingError::OpenRouterError("Missing 'splits' field for volume_split_priority output".into())
+            })?;
+
+            // Transform each {connector: [...], split} into {output: [...], split}
+            let fixed_splits: Vec<_> = splits_value
+                .as_array()
+                .ok_or_else(|| {
+                    logger::error!("decision_engine_euclid_error: 'splits' is not an array");
+                    errors::RoutingError::OpenRouterError("'splits' field must be an array".into())
+                })?
+                .iter()
+                .map(|entry| {
+                    let mut entry_map = entry.as_object().cloned().ok_or_else(|| {
+                        logger::error!("decision_engine_euclid_error: invalid split entry in volume_split_priority");
+                        errors::RoutingError::OpenRouterError("Invalid entry in splits array".into())
+                    })?;
+                    if let Some(connector) = entry_map.remove("connector") {
+                        entry_map.insert("output".to_string(), connector);
+                    }
+                    Ok::<_, error_stack::Report<errors::RoutingError>>(serde_json::Value::Object(entry_map))
+                })
+                .collect::<Result<Vec<_>, _>>()?;
+
+            let splits: Vec<VolumeSplit<Vec<ConnectorInfo>>> =
+                serde_json::from_value(serde_json::Value::Array(fixed_splits)).map_err(|e| {
+                    logger::error!(
+                        ?e,
+                        "decision_engine_euclid_error: Failed to parse volume_split_priority"
+                    );
+                    errors::RoutingError::OpenRouterError(
+                        "Failed to deserialize volume_split_priority connectors".into(),
+                    )
+                })?;
+
+            Ok(splits.into_iter().flat_map(|s| s.output).collect())
+        }
+
+        other => {
+            logger::error!(type_str=%other, "decision_engine_euclid_error: unknown output type");
+            Err(
+                errors::RoutingError::OpenRouterError(format!("Unknown output type: {other}"))
+                    .into(),
+            )
+        }
+    }
 }
 
 pub async fn create_de_euclid_routing_algo(
@@ -460,17 +710,13 @@ pub fn compare_and_log_result<T: RoutingEq<T> + Serialize>(
     result: Vec<T>,
     flow: String,
 ) {
-    let is_equal = de_result.len() == result.len()
-        && de_result
-            .iter()
-            .zip(result.iter())
-            .all(|(a, b)| T::is_equal(a, b));
-
-    if is_equal {
-        router_env::logger::info!(routing_flow=?flow, is_equal=?is_equal, "decision_engine_euclid");
-    } else {
-        router_env::logger::debug!(routing_flow=?flow, is_equal=?is_equal, de_response=?to_json_string(&de_result), hs_response=?to_json_string(&result), "decision_engine_euclid");
-    }
+    let is_equal = de_result
+        .iter()
+        .zip(result.iter())
+        .all(|(a, b)| T::is_equal(a, b));
+
+    let is_equal_in_length = de_result.len() == result.len();
+    router_env::logger::debug!(routing_flow=?flow, is_equal=?is_equal, is_equal_length=?is_equal_in_length, de_response=?to_json_string(&de_result), hs_response=?to_json_string(&result), "decision_engine_euclid");
 }
 
 pub trait RoutingEq<T> {
@@ -685,6 +931,8 @@ impl DecisionEngineErrorsInterface for or_types::ErrorResponse {
     }
 }
 
+pub type Metadata = HashMap<String, serde_json::Value>;
+
 /// Represents a single comparison condition.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -775,6 +1023,36 @@ pub struct ConnectorInfo {
     pub gateway_id: Option<String>,
 }
 
+impl TryFrom<ConnectorInfo> for DeRoutableConnectorChoice {
+    type Error = error_stack::Report<errors::RoutingError>;
+
+    fn try_from(c: ConnectorInfo) -> Result<Self, Self::Error> {
+        let gateway_id = c
+            .gateway_id
+            .map(|mca| {
+                id_type::MerchantConnectorAccountId::wrap(mca)
+                    .change_context(errors::RoutingError::GenericConversionError {
+                        from: "String".to_string(),
+                        to: "MerchantConnectorAccountId".to_string(),
+                    })
+                    .attach_printable("unable to convert MerchantConnectorAccountId from string")
+            })
+            .transpose()?;
+
+        let gateway_name = RoutableConnectors::from_str(&c.gateway_name)
+            .map_err(|_| errors::RoutingError::GenericConversionError {
+                from: "String".to_string(),
+                to: "RoutableConnectors".to_string(),
+            })
+            .attach_printable("unable to convert connector name to RoutableConnectors")?;
+
+        Ok(Self {
+            gateway_name,
+            gateway_id,
+        })
+    }
+}
+
 impl ConnectorInfo {
     pub fn new(gateway_name: String, gateway_id: Option<String>) -> Self {
         Self {
@@ -795,7 +1073,6 @@ pub enum Output {
 
 pub type Globals = HashMap<String, HashSet<ValueType>>;
 
-pub type Metadata = HashMap<String, serde_json::Value>;
 /// The program, having a default connector selection and
 /// a bunch of rules. Also can hold arbitrary metadata.
 #[derive(Clone, Debug, Serialize, Deserialize)]