diff --git a/crates/miroir-core/src/replica_selection.rs b/crates/miroir-core/src/replica_selection.rs
index f7e9a4c..97564bf 100644
--- a/crates/miroir-core/src/replica_selection.rs
+++ b/crates/miroir-core/src/replica_selection.rs
@@ -148,6 +148,21 @@ impl Default for NodeMetrics {
     }
 }
 
+/// Callback for reporting selection events.
+pub trait SelectionObserver: Send + Sync {
+    /// Called when a node is selected with its score.
+    fn report_selection(&self, node_id: &str, score: f64);
+    /// Called when exploration selects a random node.
+    fn report_exploration(&self);
+}
+
+/// No-op observer for when metrics aren't needed.
+struct NoOpObserver;
+impl SelectionObserver for NoOpObserver {
+    fn report_selection(&self, _node_id: &str, _score: f64) {}
+    fn report_exploration(&self) {}
+}
+
 /// Replica selector.
 pub struct ReplicaSelector {
     /// Configuration.
@@ -158,19 +173,27 @@ pub struct ReplicaSelector {
     rr_counter: Arc<RwLock<HashMap<String, u64>>>,
     /// Random number generator.
     rng: Arc<std::sync::Mutex<StdRng>>,
+    /// Observer for selection events.
+    observer: Arc<dyn SelectionObserver>,
 }
 
 impl ReplicaSelector {
-    /// Create a new replica selector.
-    pub fn new(config: ReplicaSelectionConfig) -> Self {
+    /// Create a new replica selector with a metrics observer.
+    pub fn new_with_observer(config: ReplicaSelectionConfig, observer: Arc<dyn SelectionObserver>) -> Self {
         Self {
             config,
             metrics: Arc::new(RwLock::new(HashMap::new())),
             rr_counter: Arc::new(RwLock::new(HashMap::new())),
             rng: Arc::new(std::sync::Mutex::new(StdRng::from_entropy())),
+            observer,
         }
     }
 
+    /// Create a new replica selector without metrics.
+    pub fn new(config: ReplicaSelectionConfig) -> Self {
+        Self::new_with_observer(config, Arc::new(NoOpObserver))
+    }
+
     /// Select a node from the given candidates.
     ///
     /// Returns the selected node ID, or None if candidates is empty.
@@ -194,12 +217,21 @@ impl ReplicaSelector {
 
         // Exploration: with probability epsilon, pick randomly
         if self.should_explore() {
-            return self.select_random(candidates);
+            self.observer.report_exploration();
+            let selected = self.select_random(candidates);
+            if let Some(ref node) = selected {
+                let score = metrics
+                    .get(node)
+                    .map(|m| m.score(&self.config))
+                    .unwrap_or(1000.0);
+                self.observer.report_selection(node.as_str(), score);
+            }
+            return selected;
         }
 
-        // Compute scores and find the minimum
-        let mut best_node = None;
+        // Compute scores and collect all nodes with the minimum score
         let mut best_score = f64::INFINITY;
+        let mut best_nodes: Vec<NodeId> = Vec::new();
 
         for node in candidates {
             let score = metrics
@@ -209,11 +241,27 @@ impl ReplicaSelector {
 
             if score < best_score {
                 best_score = score;
-                best_node = Some(node.clone());
+                best_nodes.clear();
+                best_nodes.push(node.clone());
+            } else if (score - best_score).abs() < 1e-10 {
+                // Scores are essentially equal - add to tie list
+                best_nodes.push(node.clone());
             }
         }
 
-        best_node
+        // If multiple nodes have the same best score, pick randomly
+        let selected = if best_nodes.len() == 1 {
+            best_nodes.into_iter().next()
+        } else {
+            let idx = self.rng.lock().unwrap().gen_range(0..best_nodes.len());
+            best_nodes.get(idx).cloned()
+        };
+
+        if let Some(ref node) = selected {
+            self.observer.report_selection(node.as_str(), best_score);
+        }
+
+        selected
     }
 
     /// Round-robin selection.
@@ -293,6 +341,18 @@ impl ReplicaSelector {
     }
 }
 
+impl Clone for ReplicaSelector {
+    fn clone(&self) -> Self {
+        Self {
+            config: self.config.clone(),
+            metrics: Arc::clone(&self.metrics),
+            rr_counter: Arc::clone(&self.rr_counter),
+            rng: Arc::clone(&self.rng),
+            observer: Arc::clone(&self.observer),
+        }
+    }
+}
+
 impl Default for ReplicaSelector {
     fn default() -> Self {
         Self::new(ReplicaSelectionConfig::default())
@@ -348,30 +408,9 @@ mod tests {
         let node1 = NodeId::new("node-1".to_string());
         let node2 = NodeId::new("node-2".to_string());
 
-        // Record some metrics
-        {
-            let mut metrics = selector.metrics.write().await;
-            metrics.insert(
-                node1.clone(),
-                NodeMetrics {
-                    latency_p95_ms: 10.0,
-                    in_flight: 0,
-                    error_rate: 0.0,
-                    half_life_ms: 5000,
-                    last_updated: Instant::now(),
-                },
-            );
-            metrics.insert(
-                node2.clone(),
-                NodeMetrics {
-                    latency_p95_ms: 100.0,
-                    in_flight: 0,
-                    error_rate: 0.0,
-                    half_life_ms: 5000,
-                    last_updated: Instant::now(),
-                },
-            );
-        }
+        // Seed metrics by recording successful requests
+        selector.record_success(&node1, 10.0).await;
+        selector.record_success(&node2, 100.0).await;
 
         // Should select node-1 (lower score)
         let candidates = vec![node2.clone(), node1.clone()];
@@ -416,10 +455,12 @@ mod tests {
         assert!(metrics.is_some());
         assert_eq!(metrics.unwrap().in_flight, 1);
 
+        // Record success decrements in-flight and updates latency
         selector.record_success(&node, 50.0).await;
 
         let metrics = selector.get_metrics(&node).await;
         assert!(metrics.is_some());
+        // In-flight should be decremented (from 1 to 0)
         assert_eq!(metrics.unwrap().in_flight, 0);
     }
 
diff --git a/crates/miroir-core/src/scatter.rs b/crates/miroir-core/src/scatter.rs
index 64560ec..928c541 100644
--- a/crates/miroir-core/src/scatter.rs
+++ b/crates/miroir-core/src/scatter.rs
@@ -384,11 +384,12 @@ pub struct ScatterResult {
 }
 
 #[instrument(skip_all, fields(query_seq, rf, shard_count))]
-pub fn plan_search_scatter(
+pub async fn plan_search_scatter(
     topology: &Topology,
     query_seq: u64,
     rf: usize,
     shard_count: u32,
+    replica_selector: Option<&ReplicaSelector>,
 ) -> ScatterPlan {
     let chosen_group = query_group(query_seq, topology.replica_group_count());
 
@@ -407,7 +408,16 @@ pub fn plan_search_scatter(
     let mut shard_to_node = HashMap::new();
     for shard_id in 0..shard_count {
         let replicas = crate::router::assign_shard_in_group(shard_id, group.nodes(), rf);
-        let selected = replicas[(query_seq as usize) % replicas.len()].clone();
+
+        let selected = if let Some(selector) = replica_selector {
+            match selector.select(&replicas, chosen_group).await {
+                Some(node) => node,
+                None => replicas[(query_seq as usize) % replicas.len()].clone(),
+            }
+        } else {
+            replicas[(query_seq as usize) % replicas.len()].clone()
+        };
+
         shard_to_node.insert(shard_id, selected);
     }
 
@@ -424,7 +434,7 @@ pub fn plan_search_scatter(
 ///
 /// Excludes nodes whose settings version for the given index is below `floor`.
 /// Returns None if no covering set can be assembled (caller should return 503).
-pub fn plan_search_scatter_with_version_floor(
+pub async fn plan_search_scatter_with_version_floor(
     topology: &Topology,
     query_seq: u64,
     rf: usize,
@@ -432,6 +442,7 @@ pub fn plan_search_scatter_with_version_floor(
     index: &str,
     floor: u64,
     version_checker: &impl Fn(&str, &str) -> u64,
+    replica_selector: Option<&ReplicaSelector>,
 ) -> Option<ScatterPlan> {
     let chosen_group = query_group(query_seq, topology.replica_group_count());
 
@@ -450,7 +461,7 @@ pub fn plan_search_scatter_with_version_floor(
     let mut shard_to_node = HashMap::new();
     for shard_id in 0..shard_count {
         let replicas = crate::router::assign_shard_in_group(shard_id, group.nodes(), rf);
-        // Filter by version floor, then rotate by query_seq
+        // Filter by version floor
         let eligible: Vec<_> = replicas
             .iter()
             .filter(|node_id| {
@@ -463,7 +474,17 @@ pub fn plan_search_scatter_with_version_floor(
             return None;
         }
 
-        let selected = eligible[query_seq as usize % eligible.len()];
+        let selected = if let Some(selector) = replica_selector {
+            // Convert Vec<&NodeId> to Vec<NodeId> for selector
+            let eligible_owned: Vec<NodeId> = eligible.iter().map(|&n| n.clone()).collect();
+            match selector.select(&eligible_owned, chosen_group).await {
+                Some(node) => node,
+                None => eligible[query_seq as usize % eligible.len()].clone(),
+            }
+        } else {
+            eligible[query_seq as usize % eligible.len()].clone()
+        };
+
         shard_to_node.insert(shard_id, selected.clone());
     }
 
@@ -480,12 +501,13 @@ pub fn plan_search_scatter_with_version_floor(
 ///
 /// Used when a session has a pending write and needs to read from the pinned group
 /// to ensure read-your-writes consistency.
-pub fn plan_search_scatter_for_group(
+pub async fn plan_search_scatter_for_group(
     topology: &Topology,
     query_seq: u64,
     rf: usize,
     shard_count: u32,
     pinned_group: u32,
+    replica_selector: Option<&ReplicaSelector>,
 ) -> Option<ScatterPlan> {
     let group = topology.group(pinned_group)?;
 
@@ -495,7 +517,16 @@ pub fn plan_search_scatter_for_group(
         if replicas.is_empty() {
             continue;
         }
-        let selected = replicas[query_seq as usize % replicas.len()].clone();
+
+        let selected = if let Some(selector) = replica_selector {
+            match selector.select(&replicas, pinned_group).await {
+                Some(node) => node,
+                None => replicas[query_seq as usize % replicas.len()].clone(),
+            }
+        } else {
+            replicas[query_seq as usize % replicas.len()].clone()
+        };
+
         shard_to_node.insert(shard_id, selected);
     }
 
@@ -979,43 +1010,43 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_plan_pure_function() {
+    #[tokio::test]
+    async fn test_plan_pure_function() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         assert_eq!(plan.chosen_group, 0);
         assert_eq!(plan.target_shards.len(), 64);
         assert!(plan.hedging_eligible);
     }
 
-    #[test]
-    fn test_plan_group_rotation() {
+    #[tokio::test]
+    async fn test_plan_group_rotation() {
         let topo = make_test_topology();
-        assert_eq!(plan_search_scatter(&topo, 0, 2, 64).chosen_group, 0);
-        assert_eq!(plan_search_scatter(&topo, 1, 2, 64).chosen_group, 1);
+        assert_eq!(plan_search_scatter(&topo, 0, 2, 64, None).await.chosen_group, 0);
+        assert_eq!(plan_search_scatter(&topo, 1, 2, 64, None).await.chosen_group, 1);
     }
 
-    #[test]
-    fn test_plan_shard_mapping() {
+    #[tokio::test]
+    async fn test_plan_shard_mapping() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         for s in 0..64 { assert!(plan.shard_to_node.contains_key(&s)); }
         let g0 = topo.group(0).unwrap();
         for (_, nid) in &plan.shard_to_node { assert!(g0.nodes().contains(nid)); }
     }
 
-    #[test]
-    fn test_plan_hedging() {
+    #[tokio::test]
+    async fn test_plan_hedging() {
         let mut topo = Topology::new(64, 1, 1);
         topo.add_node(Node::new(NodeId::new("n0".into()), "http://n0:7700".into(), 0));
-        assert!(!plan_search_scatter(&topo, 0, 1, 64).hedging_eligible);
-        assert!(plan_search_scatter(&make_test_topology(), 0, 2, 64).hedging_eligible);
+        assert!(!plan_search_scatter(&topo, 0, 1, 64, None).await.hedging_eligible);
+        assert!(plan_search_scatter(&make_test_topology(), 0, 2, 64, None).await.hedging_eligible);
     }
 
     #[tokio::test]
     async fn test_scatter_mock() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "doc1"}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
         let r = execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Partial).await.unwrap();
@@ -1026,7 +1057,7 @@ mod tests {
     #[tokio::test]
     async fn test_scatter_partial() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.errors.insert(NodeId::new("node-0".into()), NodeError::Timeout);
         let r = execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Partial).await.unwrap();
@@ -1036,21 +1067,21 @@ mod tests {
     #[tokio::test]
     async fn test_scatter_error_policy() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.errors.insert(NodeId::new("node-0".into()), NodeError::Timeout);
         assert!(execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Error).await.is_err());
     }
 
-    #[test]
-    fn test_plan_invalid_group() {
-        assert!(plan_search_scatter(&Topology::new(64, 0, 1), 0, 1, 64).shard_to_node.is_empty());
+    #[tokio::test]
+    async fn test_plan_invalid_group() {
+        assert!(plan_search_scatter(&Topology::new(64, 0, 1), 0, 1, 64, None).await.shard_to_node.is_empty());
     }
 
     #[tokio::test]
     async fn test_scatter_node_not_in_topo() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let r = execute_scatter(plan, &MockNodeClient::default(), make_req(), &Topology::new(64, 2, 2), UnavailableShardPolicy::Partial).await.unwrap();
         assert!(r.partial);
     }
@@ -1058,7 +1089,7 @@ mod tests {
     #[tokio::test]
     async fn test_sg_rrf() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a", "_rankingScore": 0.9}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
         let s = crate::merger::RrfStrategy::default_strategy();
@@ -1069,7 +1100,7 @@ mod tests {
     #[tokio::test]
     async fn test_sg_degraded() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a"}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
         c.errors.insert(NodeId::new("node-2".into()), NodeError::Timeout);
@@ -1105,7 +1136,7 @@ mod tests {
     #[tokio::test]
     async fn test_execute_preflight() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.preflight_responses.insert(NodeId::new("node-0".into()), PreflightResponse {
             total_docs: 30000, avg_doc_length: 50.0,
@@ -1128,7 +1159,7 @@ mod tests {
     #[tokio::test]
     async fn test_dfs_query_then_fetch() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let mut c = MockNodeClient::default();
         c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a", "_rankingScore": 0.9}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
         c.preflight_responses.insert(NodeId::new("node-0".into()), PreflightResponse {
@@ -1165,7 +1196,7 @@ mod tests {
         topo.add_node(Node::new(NodeId::new("node-1".into()), "http://node-1:7700".into(), 0));
         topo.add_node(Node::new(NodeId::new("node-2".into()), "http://node-2:7700".into(), 0));
 
-        let plan = plan_search_scatter(&topo, 0, 1, 3);
+        let plan = plan_search_scatter(&topo, 0, 1, 3, None).await;
 
         // Simulate severely skewed shard distribution
         let mut c = MockNodeClient::default();
@@ -1239,7 +1270,7 @@ mod tests {
     #[tokio::test]
     async fn test_dfs_empty_query_terms() {
         let topo = make_test_topology();
-        let plan = plan_search_scatter(&topo, 0, 2, 64);
+        let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
         let c = MockNodeClient::default();
 
         let preflight_req = PreflightRequest {
@@ -1261,7 +1292,7 @@ mod tests {
         topo.add_node(Node::new(NodeId::new("node-1".into()), "http://node-1:7700".into(), 0));
         topo.add_node(Node::new(NodeId::new("node-2".into()), "http://node-2:7700".into(), 0));
 
-        let plan = plan_search_scatter(&topo, 0, 1, 3);
+        let plan = plan_search_scatter(&topo, 0, 1, 3, None).await;
         let mut c = MockNodeClient::default();
 
         // Node 0 returns valid data
@@ -1394,7 +1425,7 @@ mod tests {
         topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
         topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
 
-        let plan = plan_search_scatter(&topo, 0, 2, 16); // query_seq=0 → group 0
+        let plan = plan_search_scatter(&topo, 0, 2, 16, None).await; // query_seq=0 → group 0
         assert_eq!(plan.chosen_group, 0);
 
         let mut c = MockNodeClient::default();
@@ -1432,7 +1463,7 @@ mod tests {
         topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
         topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
 
-        let plan = plan_search_scatter(&topo, 0, 2, 16);
+        let plan = plan_search_scatter(&topo, 0, 2, 16, None).await;
         let mut c = MockNodeClient::default();
 
         // All nodes fail
@@ -1459,7 +1490,7 @@ mod tests {
         topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
         topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
 
-        let plan = plan_search_scatter(&topo, 0, 2, 16);
+        let plan = plan_search_scatter(&topo, 0, 2, 16, None).await;
         let mut c = MockNodeClient::default();
 
         // Group 1 nodes are healthy but partial policy shouldn't use them
diff --git a/crates/miroir-core/tests/p13_3_adaptive_replica_selection.rs b/crates/miroir-core/tests/p13_3_adaptive_replica_selection.rs
new file mode 100644
index 0000000..26045d2
--- /dev/null
+++ b/crates/miroir-core/tests/p13_3_adaptive_replica_selection.rs
@@ -0,0 +1,475 @@
+//! P5.3 §13.3 Adaptive replica selection (EWMA-based) acceptance tests.
+//!
+//! Tests the EWMA-scored replica selection that replaces round-robin:
+//! - score(node) = α · latency_p95_ms + β · in_flight_count + γ · error_rate
+//! - Router picks lowest-scoring eligible node with probability 1 − ε
+//! - With ε (default 0.05) picks uniformly random for exploration
+
+use miroir_core::replica_selection::{ReplicaSelectionConfig, ReplicaSelector};
+use miroir_core::topology::{Group, NodeId};
+use std::time::Duration;
+
+/// Create a test selector with default config.
+fn test_selector() -> ReplicaSelector {
+    ReplicaSelector::new(ReplicaSelectionConfig::default())
+}
+
+/// Create a selector with custom half-life for faster test convergence.
+fn fast_test_selector(half_life_ms: u64) -> ReplicaSelector {
+    ReplicaSelector::new(ReplicaSelectionConfig {
+        ewma_half_life_ms: half_life_ms,
+        exploration_epsilon: 0.05,
+        ..Default::default()
+    })
+}
+
+/// Create a test group with 3 nodes.
+fn test_group_3() -> Group {
+    let mut group = Group::new(0);
+    group.add_node(NodeId::new("node-0".to_string()));
+    group.add_node(NodeId::new("node-1".to_string()));
+    group.add_node(NodeId::new("node-2".to_string()));
+    group
+}
+
+/// Simulate queries to measure node selection distribution.
+async fn simulate_selections(
+    selector: &ReplicaSelector,
+    candidates: &[NodeId],
+    group_id: u32,
+    count: usize,
+) -> std::collections::HashMap<String, usize> {
+    let mut distribution = std::collections::HashMap::new();
+    for _ in 0..count {
+        if let Some(node) = selector.select(candidates, group_id).await {
+            *distribution.entry(node.as_str().to_string()).or_insert(0) += 1;
+        }
+    }
+    distribution
+}
+
+// ─────────────────────────────────────────────────────────────
+// P5.3-A1: Traffic shifts away from degraded node within 2× half-life
+// ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn p5_3_a1_degraded_node_receives_less_traffic() {
+    let selector = fast_test_selector(100); // 100ms half-life for fast test
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+    let node2 = &candidates[2];
+
+    // Initialize all nodes with similar good latency (50ms)
+    selector.record_success(node0, 50.0).await;
+    selector.record_success(node1, 50.0).await;
+    selector.record_success(node2, 50.0).await;
+
+    // Run initial queries to establish baseline
+    let baseline = simulate_selections(&selector, &candidates, 0, 200).await;
+
+    // Verify roughly uniform distribution initially
+    let count0 = *baseline.get("node-0").unwrap_or(&0);
+    let count1 = *baseline.get("node-1").unwrap_or(&0);
+    let count2 = *baseline.get("node-2").unwrap_or(&0);
+
+    // Each node should get ~33% (allow 20-45% for randomness)
+    let _expected = 200 / 3;
+    assert!(
+        (20..=90).contains(&count0),
+        "node-0 baseline count {} out of expected range 20-90",
+        count0
+    );
+    assert!(
+        (20..=90).contains(&count1),
+        "node-1 baseline count {} out of expected range 20-90",
+        count1
+    );
+    assert!(
+        (20..=90).contains(&count2),
+        "node-2 baseline count {} out of expected range 20-90",
+        count2
+    );
+
+    // Induce degradation on node-1: 200ms latency
+    for _ in 0..10 {
+        selector.record_success(node1, 200.0).await;
+    }
+
+    // Wait for EWMA to propagate (2× half-life = 200ms)
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Run queries after degradation
+    let degraded = simulate_selections(&selector, &candidates, 0, 200).await;
+
+    let degraded_count1 = *degraded.get("node-1").unwrap_or(&0);
+    let degraded_count0 = *degraded.get("node-0").unwrap_or(&0);
+    let degraded_count2 = *degraded.get("node-2").unwrap_or(&0);
+
+    // Node-1 should receive significantly less traffic
+    // With latency_weight=1.0, node-1 score ~200 vs ~50 for others
+    // Expect node-1 to get <15% of traffic
+    assert!(
+        degraded_count1 < 30,
+        "degraded node-1 still receiving too much traffic: {}",
+        degraded_count1
+    );
+
+    // Healthy nodes should receive more traffic
+    assert!(
+        degraded_count0 > 50,
+        "healthy node-0 not receiving enough traffic: {}",
+        degraded_count0
+    );
+    assert!(
+        degraded_count2 > 50,
+        "healthy node-2 not receiving enough traffic: {}",
+        degraded_count2
+    );
+}
+
+// ─────────────────────────────────────────────────────────────
+// P5.3-A2: Node recovers after latency clears; distribution returns to uniform
+// ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn p5_3_a2_degraded_node_recovers() {
+    let selector = fast_test_selector(100); // 100ms half-life
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+    let node2 = &candidates[2];
+
+    // Initialize with good latency
+    selector.record_success(node0, 50.0).await;
+    selector.record_success(node1, 50.0).await;
+    selector.record_success(node2, 50.0).await;
+
+    // Degrade node-1 with 200ms latency
+    for _ in 0..10 {
+        selector.record_success(node1, 200.0).await;
+    }
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Verify degradation worked
+    let degraded_dist = simulate_selections(&selector, &candidates, 0, 100).await;
+    let degraded_count1 = *degraded_dist.get("node-1").unwrap_or(&0);
+    assert!(
+        degraded_count1 < 20,
+        "node-1 should be degraded, got {} selections",
+        degraded_count1
+    );
+
+    // Clear latency: record good responses for node-1
+    for _ in 0..10 {
+        selector.record_success(node1, 50.0).await;
+    }
+
+    // Wait for recovery (2× half-life)
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Run queries after recovery
+    let recovered = simulate_selections(&selector, &candidates, 0, 300).await;
+
+    let recovered_count0 = *recovered.get("node-0").unwrap_or(&0);
+    let recovered_count1 = *recovered.get("node-1").unwrap_or(&0);
+    let recovered_count2 = *recovered.get("node-2").unwrap_or(&0);
+
+    // All nodes should have roughly equal distribution (~1/3 each)
+    let expected = 300 / 3;
+    let tolerance = (expected as f64 * 0.4) as usize; // 40% tolerance for exploration
+
+    assert!(
+        (recovered_count1 as isize - expected as isize).abs() <= tolerance as isize,
+        "node-1 recovered count {} not close to expected {} (tolerance {})",
+        recovered_count1, expected, tolerance
+    );
+    assert!(
+        (recovered_count0 as isize - expected as isize).abs() <= tolerance as isize,
+        "node-0 count {} not close to expected {} (tolerance {})",
+        recovered_count0, expected, tolerance
+    );
+    assert!(
+        (recovered_count2 as isize - expected as isize).abs() <= tolerance as isize,
+        "node-2 count {} not close to expected {} (tolerance {})",
+        recovered_count2, expected, tolerance
+    );
+}
+
+// ─────────────────────────────────────────────────────────────
+// P5.3-A3: Exploration ensures degraded node still gets some traffic (ε = 0.05)
+//
+// With epsilon=0.05 and 3 nodes, degraded node gets ~1.7% (5%/3) of traffic
+// via random exploration. This ensures continued monitoring for recovery.
+// ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn p5_3_a3_exploration_samples_degraded_node() {
+    let selector = ReplicaSelector::new(ReplicaSelectionConfig {
+        ewma_half_life_ms: 100,
+        exploration_epsilon: 0.05, // 5% exploration
+        ..Default::default()
+    });
+
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+    let node2 = &candidates[2];
+
+    // Initialize: node-0 and node-1 healthy, node-2 severely degraded
+    for _ in 0..10 {
+        selector.record_success(node0, 50.0).await;
+        selector.record_success(node1, 50.0).await;
+        selector.record_success(node2, 500.0).await; // Very high latency
+    }
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Run 1000 queries
+    let dist = simulate_selections(&selector, &candidates, 0, 1000).await;
+
+    let count0 = *dist.get("node-0").unwrap_or(&0);
+    let count1 = *dist.get("node-1").unwrap_or(&0);
+    let count2 = *dist.get("node-2").unwrap_or(&0);
+
+    println!("Distribution: node-0={}, node-1={}, node-2={}", count0, count1, count2);
+
+    // Node-2 is severely degraded but should still get some traffic via exploration
+    // With 5% epsilon and random exploration across 3 nodes:
+    // - 5% of 1000 = 50 explorations total
+    // - 1/3 of explorations go to node-2 ≈ 17 selections
+    // Allow range 5-30 for statistical variance (3 sigma)
+    assert!(
+        (5..=30).contains(&count2),
+        "exploration not working: degraded node-2 got {} selections, expected ~17 (range 5-30)",
+        count2
+    );
+
+    // Healthy nodes should split the remaining ~95%
+    let healthy_total = count0 + count1;
+    assert!(
+        healthy_total >= 900,
+        "healthy nodes didn't get enough traffic: {}",
+        healthy_total
+    );
+
+    // Each healthy node should get roughly half of remaining
+    let expected_healthy = 475; // ~95% / 2
+    let tolerance = 100;
+    assert!(
+        (count0 as isize - expected_healthy).abs() <= tolerance,
+        "node-0 count {} not close to expected {}",
+        count0, expected_healthy
+    );
+    assert!(
+        (count1 as isize - expected_healthy).abs() <= tolerance,
+        "node-1 count {} not close to expected {}",
+        count1, expected_healthy
+    );
+}
+
+// ─────────────────────────────────────────────────────────────
+// P5.3-A4: Round-robin fallback mode works identically to Phase 1 baseline
+// ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn p5_3_a4_round_robin_fallback() {
+    let selector = ReplicaSelector::new(ReplicaSelectionConfig {
+        strategy: "round_robin".to_string(),
+        ewma_half_life_ms: 5000,
+        ..Default::default()
+    });
+
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    // Test round-robin order
+    let first = selector.select(&candidates, 0).await;
+    let second = selector.select(&candidates, 0).await;
+    let third = selector.select(&candidates, 0).await;
+    let fourth = selector.select(&candidates, 0).await;
+
+    // Should cycle through candidates in order
+    assert_eq!(first, candidates.get(0).cloned());
+    assert_eq!(second, candidates.get(1).cloned());
+    assert_eq!(third, candidates.get(2).cloned());
+    assert_eq!(fourth, candidates.get(0).cloned()); // Wrap around
+}
+
+// ─────────────────────────────────────────────────────────────
+// Additional tests for EWMA behavior
+// ─────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn ewma_converges_toward_new_values() {
+    let selector = fast_test_selector(50);
+    let group = test_group_3();
+    let node = &group.nodes()[0];
+
+    // Start with 100ms latency
+    selector.record_success(node, 100.0).await;
+    let metrics1 = selector.get_metrics(node).await;
+    assert!(metrics1.is_some());
+    let latency1 = metrics1.unwrap().latency_p95_ms;
+    assert!((90.0..=110.0).contains(&latency1));
+
+    // Shift to 50ms
+    for _ in 0..5 {
+        selector.record_success(node, 50.0).await;
+    }
+
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let metrics2 = selector.get_metrics(node).await;
+    let latency2 = metrics2.unwrap().latency_p95_ms;
+
+    // Should move toward 50
+    assert!(latency2 < latency1, "EWMA should decrease toward new value");
+    assert!(latency2 > 40.0, "EWMA shouldn't overshoot too much");
+}
+
+#[tokio::test]
+async fn in_flight_count_affects_score() {
+    let selector = test_selector();
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+
+    // Both have same latency initially
+    selector.record_success(node0, 50.0).await;
+    selector.record_success(node1, 50.0).await;
+
+    // Add in-flight requests to node-0
+    for _ in 0..5 {
+        selector.record_request_start(node0).await;
+    }
+
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    // With inflight_weight=2.0, node-0 score should be higher
+    let metrics0 = selector.get_metrics(node0).await.unwrap();
+    let metrics1 = selector.get_metrics(node1).await.unwrap();
+
+    let config = ReplicaSelectionConfig::default();
+    let score0 = metrics0.score(&config);
+    let score1 = metrics1.score(&config);
+
+    assert!(
+        score0 > score1,
+        "node-0 with in-flight requests should have higher score: {} > {}",
+        score0,
+        score1
+    );
+}
+
+#[tokio::test]
+async fn error_rate_affects_score() {
+    let selector = test_selector();
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+
+    // Both have same latency initially
+    selector.record_success(node0, 50.0).await;
+    selector.record_success(node1, 50.0).await;
+
+    // Record errors for node-0
+    for _ in 0..5 {
+        selector.record_error(node0, Some(50.0)).await;
+    }
+
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    // With error_weight=10.0, node-0 score should be higher
+    let metrics0 = selector.get_metrics(node0).await.unwrap();
+    let metrics1 = selector.get_metrics(node1).await.unwrap();
+
+    let config = ReplicaSelectionConfig::default();
+    let score0 = metrics0.score(&config);
+    let score1 = metrics1.score(&config);
+
+    assert!(
+        score0 > score1,
+        "node-0 with errors should have higher score: {} > {}",
+        score0,
+        score1
+    );
+
+    // Verify error_rate is set
+    assert!(metrics0.error_rate > 0.0);
+    assert_eq!(metrics1.error_rate, 0.0);
+}
+
+#[tokio::test]
+async fn test_exploration_epsilon_configurable() {
+    // Test with epsilon = 0.0 (no exploration)
+    let selector_no_explore = ReplicaSelector::new(ReplicaSelectionConfig {
+        exploration_epsilon: 0.0,
+        ewma_half_life_ms: 100,
+        ..Default::default()
+    });
+
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    let node0 = &candidates[0];
+    let node1 = &candidates[1];
+    let node2 = &candidates[2];
+
+    // Make node-2 severely degraded
+    selector_no_explore.record_success(node0, 50.0).await;
+    selector_no_explore.record_success(node1, 50.0).await;
+    selector_no_explore.record_success(node2, 500.0).await;
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // With no exploration, node-2 should get almost no traffic
+    let dist = simulate_selections(&selector_no_explore, &candidates, 0, 200).await;
+    let count2 = *dist.get("node-2").unwrap_or(&0);
+
+    assert_eq!(
+        count2, 0,
+        "with epsilon=0.0, degraded node should get zero traffic"
+    );
+}
+
+#[tokio::test]
+async fn test_random_strategy() {
+    let selector = ReplicaSelector::new(ReplicaSelectionConfig {
+        strategy: "random".to_string(),
+        ..Default::default()
+    });
+
+    let group = test_group_3();
+    let candidates: Vec<NodeId> = group.nodes().to_vec();
+
+    // Run many selections to test uniformity
+    let mut counts = std::collections::HashMap::new();
+    for _ in 0..1000 {
+        if let Some(node) = selector.select(&candidates, 0).await {
+            *counts.entry(node.as_str().to_string()).or_insert(0) += 1;
+        }
+    }
+
+    // Each node should get roughly 1/3 of selections
+    let expected = 1000 / 3;
+    for (node, count) in &counts {
+        let diff = (*count as isize - expected as isize).abs();
+        assert!(
+            diff <= 100, // Allow 10% variance
+            "{}: got {} selections, expected ~{}",
+            node, count, expected
+        );
+    }
+}
diff --git a/crates/miroir-proxy/src/middleware.rs b/crates/miroir-proxy/src/middleware.rs
index e2e4f6c..4fd67a8 100644
--- a/crates/miroir-proxy/src/middleware.rs
+++ b/crates/miroir-proxy/src/middleware.rs
@@ -295,6 +295,14 @@ pub struct Metrics {
     // ── §13.3 Adaptive replica selection metrics (always present) ──
     replica_selection_score: GaugeVec,
     replica_selection_exploration_total: Counter,
+
+    // ── §13.10 Idempotency metrics (always present) ──
+    idempotency_hits_total: CounterVec,
+    idempotency_cache_size: Gauge,
+
+    // ── §13.10 Query coalescing metrics (always present) ──
+    query_coalesce_subscribers_total: Counter,
+    query_coalesce_hits_total: Counter,
 }
 
 impl Clone for Metrics {
@@ -386,6 +394,10 @@ impl Clone for Metrics {
             antientropy_last_scan_completed_seconds: self.antientropy_last_scan_completed_seconds.clone(),
             replica_selection_score: self.replica_selection_score.clone(),
             replica_selection_exploration_total: self.replica_selection_exploration_total.clone(),
+            idempotency_hits_total: self.idempotency_hits_total.clone(),
+            idempotency_cache_size: self.idempotency_cache_size.clone(),
+            query_coalesce_subscribers_total: self.query_coalesce_subscribers_total.clone(),
+            query_coalesce_hits_total: self.query_coalesce_hits_total.clone(),
         }
     }
 }
@@ -966,6 +978,27 @@ impl Metrics {
         reg!(replica_selection_score);
         reg!(replica_selection_exploration_total);
 
+        // ── §13.10 Idempotency metrics (always present) ──
+        let idempotency_hits_total = CounterVec::new(
+            Opts::new("miroir_idempotency_hits_total", "Idempotency key hits by outcome"),
+            &["outcome"],
+        ).expect("create idempotency_hits_total");
+        let idempotency_cache_size = Gauge::with_opts(
+            Opts::new("miroir_idempotency_cache_size", "Current number of entries in idempotency cache")
+        ).expect("create idempotency_cache_size");
+        reg!(idempotency_hits_total);
+        reg!(idempotency_cache_size);
+
+        // ── §13.10 Query coalescing metrics (always present) ──
+        let query_coalesce_subscribers_total = Counter::with_opts(
+            Opts::new("miroir_query_coalesce_subscribers_total", "Total number of subscribers to coalesced queries")
+        ).expect("create query_coalesce_subscribers_total");
+        let query_coalesce_hits_total = Counter::with_opts(
+            Opts::new("miroir_query_coalesce_hits_total", "Total number of queries that hit an in-flight coalesced query")
+        ).expect("create query_coalesce_hits_total");
+        reg!(query_coalesce_subscribers_total);
+        reg!(query_coalesce_hits_total);
+
         Self {
             registry,
             request_duration,
@@ -1053,6 +1086,10 @@ impl Metrics {
             antientropy_last_scan_completed_seconds,
             replica_selection_score,
             replica_selection_exploration_total,
+            idempotency_hits_total,
+            idempotency_cache_size,
+            query_coalesce_subscribers_total,
+            query_coalesce_hits_total,
         }
     }
 
@@ -1763,6 +1800,26 @@ impl Metrics {
         self.replica_selection_exploration_total.inc();
     }
 
+    // ── §13.10 Idempotency metrics ──
+
+    pub fn inc_idempotency_hit(&self, outcome: &str) {
+        self.idempotency_hits_total.with_label_values(&[outcome]).inc();
+    }
+
+    pub fn set_idempotency_cache_size(&self, size: u64) {
+        self.idempotency_cache_size.set(size as f64);
+    }
+
+    // ── §13.10 Query coalescing metrics ──
+
+    pub fn inc_query_coalesce_subscribers(&self, count: u64) {
+        self.query_coalesce_subscribers_total.inc_by(count as f64);
+    }
+
+    pub fn inc_query_coalesce_hits(&self) {
+        self.query_coalesce_hits_total.inc();
+    }
+
     pub fn registry(&self) -> &Registry {
         &self.registry
     }
diff --git a/crates/miroir-proxy/src/routes/admin_endpoints.rs b/crates/miroir-proxy/src/routes/admin_endpoints.rs
index 16613d9..2f6e571 100644
--- a/crates/miroir-proxy/src/routes/admin_endpoints.rs
+++ b/crates/miroir-proxy/src/routes/admin_endpoints.rs
@@ -12,6 +12,7 @@ use miroir_core::{
     migration::{MigrationConfig, MigrationCoordinator},
     rebalancer::{MigrationExecutor, Rebalancer, RebalancerConfig, RebalancerMetrics},
     rebalancer_worker::{RebalancerMetricsCallback, RebalancerWorker, RebalancerWorkerConfig, TopologyChangeEvent},
+ replica_selection::{ReplicaSelector, SelectionObserver},
     router,
     scatter::{DeleteByFilterRequest, FetchDocumentsRequest, FetchDocumentsResponse, WriteRequest},
     task_registry::TaskRegistryImpl,
@@ -301,6 +302,23 @@ fn now_ms() -> i64 {
         .as_millis() as i64
 }
 
+/// Metrics observer for replica selection events.
+///
+/// Reports selection scores and exploration events to Prometheus.
+struct ReplicaSelectionMetricsObserver {
+    metrics: super::super::middleware::Metrics,
+}
+
+impl SelectionObserver for ReplicaSelectionMetricsObserver {
+    fn report_selection(&self, node_id: &str, score: f64) {
+        self.metrics.set_replica_selection_score(node_id, score);
+    }
+
+    fn report_exploration(&self) {
+        self.metrics.inc_replica_selection_exploration();
+    }
+}
+
 /// Shared application state for admin endpoints.
 #[derive(Clone)]
 pub struct AppState {
@@ -338,6 +356,10 @@ pub struct AppState {
     pub mode_c_worker: Option<Arc<ModeCWorker>>,
     /// Adaptive replica selector (plan §13.3).
     pub replica_selector: Arc<miroir_core::replica_selection::ReplicaSelector>,
+    /// Idempotency cache for write deduplication (plan §13.10).
+    pub idempotency_cache: Arc<miroir_core::idempotency::IdempotencyCache>,
+    /// Query coalescer for read deduplication (plan §13.10).
+    pub query_coalescer: Arc<miroir_core::idempotency::QueryCoalescer>,
 }
 
 impl AppState {
@@ -633,7 +655,22 @@ impl AppState {
             alias_registry,
             leader_election,
             mode_c_worker,
-            replica_selector: Arc::new(miroir_core::replica_selection::ReplicaSelector::default()),
+            replica_selector: {
+                let config = config.replica_selection.clone();
+                let observer = Arc::new(ReplicaSelectionMetricsObserver {
+                    metrics: metrics.clone(),
+                });
+                Arc::new(ReplicaSelector::new_with_observer(config, observer))
+            },
+            idempotency_cache: Arc::new(miroir_core::idempotency::IdempotencyCache::new(
+                config.idempotency.max_cached_keys as usize,
+                config.idempotency.ttl_seconds,
+            )),
+            query_coalescer: Arc::new(miroir_core::idempotency::QueryCoalescer::new(
+                config.query_coalescing.window_ms,
+                config.query_coalescing.max_pending_queries as usize,
+                config.query_coalescing.max_subscribers as usize,
+            )),
         }
     }
 
diff --git a/crates/miroir-proxy/src/routes/multi_search.rs b/crates/miroir-proxy/src/routes/multi_search.rs
index 7247452..241fee3 100644
--- a/crates/miroir-proxy/src/routes/multi_search.rs
+++ b/crates/miroir-proxy/src/routes/multi_search.rs
@@ -29,6 +29,7 @@ pub struct MultiSearchState {
     pub node_master_key: String,
     pub metrics: crate::middleware::Metrics,
     pub alias_registry: Arc<miroir_core::alias::AliasRegistry>,
+    pub replica_selector: Arc<miroir_core::replica_selection::ReplicaSelector>,
 }
 
 /// Multi-search request (plan §13.11).
@@ -248,17 +249,27 @@ where
             let config = state.config.clone();
             let strategy = strategy.clone();
             let policy = policy;
+            let replica_selector = state.replica_selector.clone();
 
             async move {
                 let start = Instant::now();
 
+                // Determine if we should use adaptive selection
+                let use_adaptive = config.replica_selection.strategy == "adaptive";
+                let replica_selector_ref = if use_adaptive {
+                    Some(replica_selector.as_ref())
+                } else {
+                    None
+                };
+
                 // Plan scatter for this query
                 let plan = plan_search_scatter(
                     &topology,
                     0,
                     config.replication_factor as usize,
                     config.shards,
-                );
+                    replica_selector_ref,
+                ).await;
 
                 // Build search request
                 let filter_value = query.filter.as_ref()
diff --git a/crates/miroir-proxy/src/routes/search.rs b/crates/miroir-proxy/src/routes/search.rs
index 466a8a3..a4fb0da 100644
--- a/crates/miroir-proxy/src/routes/search.rs
+++ b/crates/miroir-proxy/src/routes/search.rs
@@ -6,7 +6,9 @@ use axum::response::Response;
 use axum::Json;
 use miroir_core::api_error::{MeilisearchError, MiroirCode};
 use miroir_core::config::UnavailableShardPolicy;
+use miroir_core::idempotency::{QueryFingerprint, canonicalize_json};
 use miroir_core::merger::ScoreMergeStrategy;
+use miroir_core::replica_selection::SelectionObserver;
 use miroir_core::scatter::{
     dfs_query_then_fetch_search, plan_search_scatter, plan_search_scatter_for_group, plan_search_scatter_with_version_floor, SearchRequest, NodeClient,
 };
@@ -14,12 +16,29 @@ use miroir_core::session_pinning::WaitStrategy;
 use serde::Deserialize;
 use serde_json::Value;
 use std::sync::Arc;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use tracing::{debug, error, info, info_span, instrument, warn};
 
 use crate::middleware::SessionId;
 use crate::routes::admin_endpoints::{AppState, parse_rate_limit};
 
+/// Metrics observer for replica selection events.
+///
+/// Reports selection scores and exploration events to Prometheus.
+struct MetricsObserver {
+    metrics: crate::middleware::Metrics,
+}
+
+impl SelectionObserver for MetricsObserver {
+    fn report_selection(&self, node_id: &str, score: f64) {
+        self.metrics.set_replica_selection_score(node_id, score);
+    }
+
+    fn report_exploration(&self) {
+        self.metrics.inc_replica_selection_exploration();
+    }
+}
+
 /// Hash a value for logging (obfuscates sensitive data like IPs).
 fn hash_for_log(value: &str) -> String {
     use std::hash::{Hash, Hasher};
@@ -35,11 +54,16 @@ fn hash_for_log(value: &str) -> String {
 pub struct ProxyNodeClient {
     client: Arc<crate::client::HttpClient>,
     metrics: crate::middleware::Metrics,
+    replica_selector: Option<Arc<miroir_core::replica_selection::ReplicaSelector>>,
 }
 
 impl ProxyNodeClient {
-    pub fn new(client: Arc<crate::client::HttpClient>, metrics: crate::middleware::Metrics) -> Self {
-        Self { client, metrics }
+    pub fn new(
+        client: Arc<crate::client::HttpClient>,
+        metrics: crate::middleware::Metrics,
+        replica_selector: Option<Arc<miroir_core::replica_selection::ReplicaSelector>>,
+    ) -> Self {
+        Self { client, metrics, replica_selector }
     }
 }
 
@@ -286,6 +310,78 @@ async fn search_handler(
         );
     }
 
+    // Query coalescing (plan §13.10): Check for identical in-flight queries
+    // Skip for multi-target aliases (each target is different)
+    if state.config.query_coalescing.enabled && resolved_targets.len() == 1 {
+        // Build fingerprint from canonicalized query body + index + settings version
+        let settings_version = state.settings_broadcast.current_version().await;
+        let query_body = serde_json::to_value(&body).unwrap_or(Value::Null);
+        let fingerprint = QueryFingerprint::new(effective_index.clone(), &query_body, settings_version);
+
+        // Try to coalesce with an existing in-flight query
+        if let Some(mut rx) = state.query_coalescer.try_coalesce(fingerprint.clone()).await {
+            // Successfully subscribed to an in-flight query - wait for its result
+            state.metrics.inc_query_coalesce_hits();
+            debug!(
+                index = %effective_index,
+                "query coalesced: waiting for in-flight query result"
+            );
+
+            // Wait for the response (with timeout matching the scatter timeout)
+            let timeout = Duration::from_millis(state.config.scatter.node_timeout_ms);
+            let response_bytes = tokio::time::timeout(timeout, async move {
+                rx.recv().await
+            }).await;
+
+            match response_bytes {
+                Ok(Ok(response_bytes)) => {
+                    // Received response from coalesced query
+                    let response_body: Value = match serde_json::from_slice(&response_bytes) {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!(error = %e, "failed to deserialize coalesced query response");
+                            return Err(StatusCode::INTERNAL_SERVER_ERROR);
+                        }
+                    };
+
+                    // Build response with appropriate headers
+                    let mut response = Response::builder()
+                        .status(StatusCode::OK)
+                        .header("content-type", "application/json");
+
+                    // Add settings headers
+                    if state.settings_broadcast.is_in_flight(&effective_index).await {
+                        response = response.header("X-Miroir-Settings-Inconsistent", "true");
+                    }
+                    if settings_version > 0 {
+                        response = response.header("X-Miroir-Settings-Version", settings_version.to_string());
+                    }
+
+                    let response = response
+                        .body(axum::body::Body::from(serde_json::to_string(&response_body).unwrap()))
+                        .unwrap();
+
+                    tracing::info!(
+                        target: "miroir.search_coalesced",
+                        index = %effective_index,
+                        duration_ms = start.elapsed().as_millis() as u64,
+                        "coalesced search completed"
+                    );
+
+                    return Ok(response);
+                }
+                Ok(Err(_)) => {
+                    // Channel closed without receiving response - proceed with normal scatter
+                    debug!("coalesced query channel closed, proceeding with normal scatter");
+                }
+                Err(_) => {
+                    // Timeout waiting for coalesced query - proceed with normal scatter
+                    debug!("timeout waiting for coalesced query, proceeding with normal scatter");
+                }
+            }
+        }
+    }
+
     // Handle multi-target alias fanout (plan §13.7, §13.11, §13.17)
     // Multi-target aliases (ILM read_alias) require fanning out to all targets
     // and merging results by _rankingScore
@@ -348,8 +444,17 @@ async fn search_handler(
             rf = state.config.replication_factor,
             min_settings_version,
             pinned_group = ?pinned_group,
+            strategy = %state.config.replica_selection.strategy,
         ).entered();
 
+        // Determine if we should use adaptive selection
+        let use_adaptive = state.config.replica_selection.strategy == "adaptive";
+        let replica_selector_ref = if use_adaptive {
+            Some(state.replica_selector.as_ref())
+        } else {
+            None
+        };
+
         // Session pinning: if pinned_group is set, use group-specific planning
         if let Some(group) = pinned_group {
             match plan_search_scatter_for_group(
@@ -358,7 +463,8 @@ async fn search_handler(
                 state.config.replication_factor as usize,
                 state.config.shards,
                 group,
-            ) {
+                replica_selector_ref,
+            ).await {
                 Some(p) => p,
                 None => {
                     // Pinned group not available - fall back to normal planning
@@ -367,7 +473,7 @@ async fn search_handler(
                         pinned_group = group,
                         "pinned group unavailable, falling back to normal routing"
                     );
-                    plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards)
+                    plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards, replica_selector_ref).await
                 }
             }
         } else if let Some(floor) = min_settings_version {
@@ -392,7 +498,8 @@ async fn search_handler(
                         })
                     })
                 },
-            );
+                replica_selector_ref,
+            ).await;
 
             match plan_result {
                 Some(p) => p,
@@ -410,7 +517,7 @@ async fn search_handler(
             }
         } else {
             // No version floor requested, use normal planning
-            plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards)
+            plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards, replica_selector_ref).await
         }
     };
     let node_count = plan.shard_to_node.len() as u64;
@@ -441,6 +548,28 @@ async fn search_handler(
     // Use score-based merge strategy (OP#4: requires global IDF)
     let strategy = ScoreMergeStrategy::new();
 
+    // Register for query coalescing (plan §13.10) - after try_coalesce, before scatter
+    // Only register if coalescing is enabled and this is a single-target query
+    let (tx, fingerprint) = if state.config.query_coalescing.enabled && resolved_targets.len() == 1 {
+        let settings_version = state.settings_broadcast.current_version().await;
+        let query_body = serde_json::to_value(&body).unwrap_or(Value::Null);
+        let fp = QueryFingerprint::new(effective_index.clone(), &query_body, settings_version);
+
+        match state.query_coalescer.register(fp.clone()).await {
+            Ok(broadcast_tx) => {
+                state.metrics.inc_query_coalesce_subscribers(1); // First subscriber = the query itself
+                (Some(broadcast_tx), Some(fp))
+            }
+            Err(_) => {
+                // Failed to register (too many pending queries) - proceed without coalescing
+                debug!("too many pending queries, proceeding without coalescing registration");
+                (None, None)
+            }
+        }
+    } else {
+        (None, None)
+    };
+
     // Execute DFS query-then-fetch
     let mut result = dfs_query_then_fetch_search(
         plan,
@@ -465,17 +594,36 @@ async fn search_handler(
     }
 
     // Build response body
-    let mut body = serde_json::json!({
+    let response_body = serde_json::json!({
         "hits": result.hits,
         "estimatedTotalHits": result.estimated_total_hits,
         "processingTimeMs": result.processing_time_ms,
     });
 
     // Only include facetDistribution if facets were requested
+    let mut body = response_body.clone();
     if let Some(facets) = &result.facet_distribution {
         body["facetDistribution"] = serde_json::to_value(facets).unwrap_or(Value::Null);
     }
 
+    // Broadcast result to coalesced queries (plan §13.10)
+    if let (Some(broadcast_tx), Some(fp)) = (tx, fingerprint) {
+        let response_bytes = serde_json::to_vec(&body).unwrap_or_default();
+        let subscriber_count = broadcast_tx.receiver_count();
+        if subscriber_count > 1 {
+            // There are other queries waiting for this result - broadcast to them
+            state.metrics.inc_query_coalesce_subscribers(subscriber_count as u64 - 1);
+            let _ = broadcast_tx.send(response_bytes.clone());
+            debug!(
+                index = %effective_index,
+                subscribers = subscriber_count - 1,
+                "broadcast search result to coalesced queries"
+            );
+        }
+        // Unregister the query after broadcasting
+        state.query_coalescer.unregister(&fp).await;
+    }
+
     // Build response with optional headers
     let mut response = Response::builder()
         .status(StatusCode::OK)