Add Live reconciliation coverage

- Cover live exec-event pre-dispatch fill and batch paths
- Assert clock factory and exec-manager config mapping
- Check open-only missing orders keep submitted state

Author: cjdsellers

crates/live/src/node/config.rs

@@ -1311,6 +1311,95 @@ mod tests {
         assert_eq!(converted.purge_account_events_lookback_mins, Some(3));
     }
 
+    #[rstest]
+    fn test_live_exec_engine_config_converts_to_execution_manager_config() {
+        let config = LiveExecEngineConfig {
+            reconciliation: false,
+            reconciliation_lookback_mins: Some(45),
+            reconciliation_instrument_ids: Some(vec![
+                "ETHUSDT.BINANCE".to_string(),
+                "BTCUSDT.BINANCE".to_string(),
+            ]),
+            filter_unclaimed_external_orders: true,
+            filter_position_reports: true,
+            filtered_client_order_ids: Some(vec!["O-001".to_string(), "O-002".to_string()]),
+            generate_missing_orders: false,
+            inflight_check_interval_ms: 321,
+            inflight_check_threshold_ms: 654,
+            inflight_check_retries: 7,
+            open_check_interval_secs: Some(1.5),
+            open_check_lookback_mins: Some(9),
+            open_check_threshold_ms: 234,
+            open_check_missing_retries: 4,
+            open_check_open_only: false,
+            max_single_order_queries_per_cycle: 8,
+            single_order_query_delay_ms: 76,
+            position_check_interval_secs: Some(2.5),
+            position_check_lookback_mins: 11,
+            position_check_threshold_ms: 345,
+            position_check_retries: 6,
+            purge_closed_orders_buffer_mins: Some(12),
+            purge_closed_positions_buffer_mins: Some(13),
+            purge_account_events_lookback_mins: Some(14),
+            purge_from_database: true,
+            ..Default::default()
+        };
+
+        let converted = ExecutionManagerConfig::from(&config);
+
+        assert!(!converted.reconciliation);
+        assert_eq!(converted.lookback_mins, Some(45));
+        assert_eq!(converted.reconciliation_instrument_ids.len(), 2);
+        assert!(
+            converted
+                .reconciliation_instrument_ids
+                .contains(&InstrumentId::from("ETHUSDT.BINANCE"))
+        );
+        assert!(
+            converted
+                .reconciliation_instrument_ids
+                .contains(&InstrumentId::from("BTCUSDT.BINANCE"))
+        );
+        assert!(converted.filter_unclaimed_external);
+        assert!(converted.filter_position_reports);
+        assert_eq!(converted.filtered_client_order_ids.len(), 2);
+        assert!(
+            converted
+                .filtered_client_order_ids
+                .contains(&ClientOrderId::from("O-001"))
+        );
+        assert!(
+            converted
+                .filtered_client_order_ids
+                .contains(&ClientOrderId::from("O-002"))
+        );
+        assert!(!converted.generate_missing_orders);
+        assert_eq!(converted.inflight_check_interval_ms, 321);
+        assert_eq!(converted.inflight_threshold_ms, 654);
+        assert_eq!(converted.inflight_max_retries, 7);
+        assert_eq!(converted.open_check_interval_secs, Some(1.5));
+        assert_eq!(converted.open_check_lookback_mins, Some(9));
+        assert_eq!(
+            converted.open_check_threshold_ns,
+            234 * NANOSECONDS_IN_MILLISECOND
+        );
+        assert_eq!(converted.open_check_missing_retries, 4);
+        assert!(!converted.open_check_open_only);
+        assert_eq!(converted.max_single_order_queries_per_cycle, 8);
+        assert_eq!(converted.single_order_query_delay_ms, 76);
+        assert_eq!(converted.position_check_interval_secs, Some(2.5));
+        assert_eq!(converted.position_check_lookback_mins, 11);
+        assert_eq!(
+            converted.position_check_threshold_ns,
+            345 * NANOSECONDS_IN_MILLISECOND
+        );
+        assert_eq!(converted.position_check_retries, 6);
+        assert_eq!(converted.purge_closed_orders_buffer_mins, Some(12));
+        assert_eq!(converted.purge_closed_positions_buffer_mins, Some(13));
+        assert_eq!(converted.purge_account_events_lookback_mins, Some(14));
+        assert!(converted.purge_from_database);
+    }
+
     #[rstest]
     fn test_live_risk_engine_config_converts_to_risk_engine_config() {
         let config = LiveRiskEngineConfig {

crates/live/src/node/mod.rs

@@ -1157,56 +1157,15 @@ impl LiveNode {
                         residual_events += 1;
                     }
 
-                    let mut close_ids: Vec<ClientOrderId> = Vec::new();
-
-                    match &evt {
-                        ExecutionEvent::Order(order_evt) => {
-                            self.exec_manager.observe_order_event(order_evt);
-                            close_ids.push(order_evt.client_order_id());
-                        }
-                        ExecutionEvent::OrderSubmittedBatch(batch) => {
-                            for submitted in &batch.events {
-                                self.exec_manager.record_local_activity(submitted.client_order_id);
-                            }
-                        }
-                        ExecutionEvent::OrderAcceptedBatch(batch) => {
-                            for accepted in &batch.events {
-                                // Stamp after clearing: `clear_recon_tracking` drops the
-                                // local-activity mark, the missing-order grace gate.
-                                self.exec_manager.clear_recon_tracking(
-                                    &accepted.client_order_id, true,
-                                );
-                                self.exec_manager.record_local_activity(accepted.client_order_id);
-                            }
-                        }
-                        ExecutionEvent::OrderCanceledBatch(batch) => {
-                            for canceled in &batch.events {
-                                self.exec_manager.clear_recon_tracking(
-                                    &canceled.client_order_id, true,
-                                );
-                                self.exec_manager.record_local_activity(canceled.client_order_id);
-                                close_ids.push(canceled.client_order_id);
-                            }
-                        }
-                        ExecutionEvent::Report(report) => {
-                            if let ExecutionReport::Fill(fill_report) = report
-                                && self.exec_manager.is_fill_recently_processed(&fill_report.trade_id) {
-                                    log::debug!(
-                                        "Skipping recently processed fill report: {}",
-                                        fill_report.trade_id,
-                                    );
-                                    record_runner_dispatch(
-                                        &metrics,
-                                        RunnerMetricChannel::ExecEvents,
-                                        dispatch_start,
-                                        metrics_start,
-                                    );
-                                    continue;
-                                }
-                            self.exec_manager.observe_execution_report(report);
-                        }
-                        ExecutionEvent::Account(_) => {}
-                    }
+                    let Some(close_ids) = self.observe_exec_event_before_dispatch(&evt) else {
+                        record_runner_dispatch(
+                            &metrics,
+                            RunnerMetricChannel::ExecEvents,
+                            dispatch_start,
+                            metrics_start,
+                        );
+                        continue;
+                    };
 
                     AsyncRunner::handle_exec_event(evt);
 
@@ -1504,6 +1463,60 @@ impl LiveNode {
         }
     }
 
+    fn observe_exec_event_before_dispatch(
+        &mut self,
+        evt: &ExecutionEvent,
+    ) -> Option<Vec<ClientOrderId>> {
+        let mut close_ids = Vec::new();
+
+        match evt {
+            ExecutionEvent::Order(order_evt) => {
+                self.exec_manager.observe_order_event(order_evt);
+                close_ids.push(order_evt.client_order_id());
+            }
+            ExecutionEvent::OrderSubmittedBatch(batch) => {
+                for submitted in &batch.events {
+                    self.exec_manager
+                        .record_local_activity(submitted.client_order_id);
+                }
+            }
+            ExecutionEvent::OrderAcceptedBatch(batch) => {
+                for accepted in &batch.events {
+                    self.exec_manager
+                        .clear_recon_tracking(&accepted.client_order_id, true);
+                    self.exec_manager
+                        .record_local_activity(accepted.client_order_id);
+                }
+            }
+            ExecutionEvent::OrderCanceledBatch(batch) => {
+                for canceled in &batch.events {
+                    self.exec_manager
+                        .clear_recon_tracking(&canceled.client_order_id, true);
+                    self.exec_manager
+                        .record_local_activity(canceled.client_order_id);
+                    close_ids.push(canceled.client_order_id);
+                }
+            }
+            ExecutionEvent::Report(report) => {
+                if let ExecutionReport::Fill(fill_report) = report
+                    && self
+                        .exec_manager
+                        .is_fill_recently_processed(&fill_report.trade_id)
+                {
+                    log::debug!(
+                        "Skipping recently processed fill report: {}",
+                        fill_report.trade_id,
+                    );
+                    return None;
+                }
+                self.exec_manager.observe_execution_report(report);
+            }
+            ExecutionEvent::Account(_) => {}
+        }
+
+        Some(close_ids)
+    }
+
     /// Gets the node's environment.
     #[must_use]
     pub fn environment(&self) -> Environment {
@@ -2305,7 +2318,7 @@ mod tests {
     use nautilus_common::{
         actor::DataActor,
         cache::Cache,
-        clock::Clock,
+        clock::{Clock, TestClock},
         enums::SerializationEncoding,
         messages::execution::{SubmitOrder, TradingCommand},
         msgbus::{
@@ -2321,8 +2334,10 @@ mod tests {
     use nautilus_model::{
         data::QuoteTick,
         enums::{OmsType, OrderStatus, OrderType},
+        events::{OrderAcceptedBatch, order::spec::OrderAcceptedSpec},
         identifiers::{
-            AccountId, ClientId, InstrumentId, PositionId, StrategyId, TraderId, VenueOrderId,
+            AccountId, ClientId, InstrumentId, PositionId, StrategyId, TradeId, TraderId,
+            VenueOrderId,
         },
         instruments::{Instrument, InstrumentAny, stubs::crypto_perpetual_ethusdt},
         orders::{OrderTestBuilder, stubs::TestOrderEventStubs},
@@ -2363,6 +2378,100 @@ mod tests {
         assert_eq!(output, expected);
     }
 
+    #[rstest]
+    fn test_observe_exec_event_before_dispatch_skips_recent_fill_report() {
+        let config = LiveNodeConfig {
+            exec_engine: crate::config::LiveExecEngineConfig {
+                reconciliation: false,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let mut node = LiveNode::build("FillSkipNode".to_string(), Some(config)).unwrap();
+        let event = stub_exec_event();
+        let trade_id = TradeId::from("T-001");
+
+        let close_ids = node.observe_exec_event_before_dispatch(&event);
+        assert_eq!(close_ids, Some(Vec::new()));
+        assert!(!node.exec_manager.is_fill_recently_processed(&trade_id));
+
+        node.exec_manager.mark_fill_processed(trade_id);
+
+        let close_ids = node.observe_exec_event_before_dispatch(&event);
+        assert_eq!(close_ids, None);
+    }
+
+    #[rstest]
+    fn test_observe_exec_event_before_dispatch_accepted_batch_stamps_local_activity() {
+        let config = LiveNodeConfig {
+            exec_engine: crate::config::LiveExecEngineConfig {
+                reconciliation: true,
+                open_check_threshold_ms: 5_000,
+                single_order_query_delay_ms: 0,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let mut node = LiveNode::build("AcceptedBatchNode".to_string(), Some(config)).unwrap();
+        let account_id = AccountId::from("TEST-ACCEPTED-BATCH-001");
+        let client_id = ClientId::from("TEST-ACCEPTED-BATCH");
+        let instrument = crypto_perpetual_ethusdt();
+        let instrument_id = instrument.id();
+        let client_order_id = ClientOrderId::from("O-ACCEPTED-BATCH");
+        let venue_order_id = VenueOrderId::from("V-ACCEPTED-BATCH");
+
+        node.kernel
+            .cache
+            .borrow_mut()
+            .add_instrument(InstrumentAny::CryptoPerpetual(instrument))
+            .unwrap();
+        insert_accepted_limit_order_in_node(
+            &node,
+            account_id,
+            client_id,
+            instrument_id,
+            client_order_id,
+            venue_order_id,
+        );
+
+        assert_eq!(node.exec_manager.check_open_order_queries().len(), 1);
+
+        let accepted = OrderAcceptedSpec::builder()
+            .instrument_id(instrument_id)
+            .client_order_id(client_order_id)
+            .venue_order_id(venue_order_id)
+            .account_id(account_id)
+            .build();
+        let event = ExecutionEvent::OrderAcceptedBatch(OrderAcceptedBatch::new(vec![accepted]));
+
+        let close_ids = node.observe_exec_event_before_dispatch(&event);
+
+        assert_eq!(close_ids, Some(Vec::new()));
+        assert!(node.exec_manager.check_open_order_queries().is_empty());
+    }
+
+    #[rstest]
+    fn test_live_node_builder_clock_factory_drives_kernel_clock() {
+        let calls = Rc::new(Cell::new(0usize));
+        let calls_in_factory = calls.clone();
+        let sentinel = UnixNanos::from(123_456_789_u64);
+
+        let node = LiveNode::builder(TraderId::from("TESTER-001"), Environment::Sandbox)
+            .unwrap()
+            .with_reconciliation(false)
+            .with_clock_factory(move || {
+                calls_in_factory.set(calls_in_factory.get() + 1);
+                let mut clock = TestClock::new();
+                clock.advance_time(sentinel, true);
+                Rc::new(RefCell::new(clock)) as Rc<RefCell<dyn Clock>>
+            })
+            .build()
+            .unwrap();
+
+        assert_eq!(node.kernel().clock().borrow().timestamp_ns(), sentinel);
+        assert_eq!(calls.get(), 1);
+    }
+
     #[derive(Debug)]
     struct ReplayKernelEventStore {
         fail_restore: bool,

crates/live/tests/manager.rs

@@ -7617,6 +7617,39 @@ async fn test_check_open_orders_submitted_missing_at_venue_generates_rejected()
     }
 }
 
+#[rstest]
+#[tokio::test]
+async fn test_check_open_orders_open_only_missing_venue_order_does_not_reject() {
+    let config = ExecutionManagerConfig {
+        open_check_threshold_ns: 0,
+        open_check_missing_retries: 1,
+        open_check_open_only: true,
+        ..Default::default()
+    };
+    let mut ctx = TestContext::with_config(config);
+    ctx.add_instrument(test_instrument());
+
+    let order = create_limit_order(
+        "O-OPEN-ONLY",
+        test_instrument_id(),
+        OrderSide::Buy,
+        "10.0",
+        "100.0",
+    );
+    let submitted = TestOrderEventStubs::submitted(&order, test_account_id());
+    ctx.add_order(order);
+    ctx.cache.borrow_mut().update_order(&submitted).unwrap();
+
+    let mock_client = MockExecutionClient::new(vec![]);
+    let clients: Vec<&dyn ExecutionClient> = vec![&mock_client];
+
+    let events = ctx.manager.check_open_orders(&clients).await;
+
+    assert!(events.is_empty());
+    let cached_order = ctx.get_order(&ClientOrderId::from("O-OPEN-ONLY")).unwrap();
+    assert_eq!(cached_order.status(), OrderStatus::Submitted);
+}
+
 #[cfg_attr(
     not(all(feature = "simulation", madsim)),
     tokio::test(start_paused = true)