将手工观察接入执行时钟并保留来源与账本语义

crates/fidc-core/src/manual_execution.rs

@@ -413,7 +413,7 @@ pub struct AppliedManualFill {
 /// One replay owns its immutable trace and progress. Advancing is atomic even
 /// if a later receipt in the same step disagrees with the shadow account.
 pub struct ManualReplayCursor {
-    replay: ManualExecutionReplay,
+    replay: std::sync::Arc<ManualExecutionReplay>,
     indices: Vec<(usize, usize, usize)>,
     cursor: usize,
     clock: Option<DateTime<Utc>>,
@@ -448,6 +448,10 @@ pub struct ManualReplayApplication {
 
 impl ManualReplayCursor {
     pub fn new(replay: ManualExecutionReplay) -> Result<Self, String> {
+        Self::from_shared(std::sync::Arc::new(replay))
+    }
+
+    pub fn from_shared(replay: std::sync::Arc<ManualExecutionReplay>) -> Result<Self, String> {
         replay.validate()?;
         let mut indices = Vec::new();
         for (a, action) in replay.actions.iter().enumerate() {
@@ -485,6 +489,27 @@ impl ManualReplayCursor {
         portfolio: &mut PortfolioState,
         data: &DataSet,
         has_pending_orders: bool,
+    ) -> Result<Vec<ManualReplayApplication>, String> {
+        let end = self.cursor
+            + self.indices[self.cursor..].iter().take_while(|&&(a, o, f)| {
+                self.replay.actions[a].orders[o].fills[f].observed_at <= at
+            }).count();
+        self.advance_through(at, end, portfolio, data, has_pending_orders)
+    }
+
+    /// One receipt at a time lets callbacks observe the intermediate state
+    /// when multiple fills share a timestamp but have distinct durable sequences.
+    pub fn advance_next(
+        &mut self, portfolio: &mut PortfolioState, data: &DataSet, has_pending_orders: bool,
+    ) -> Result<Option<ManualReplayApplication>, String> {
+        let Some(at) = self.next_observation_at() else { return Ok(None); };
+        let mut applications = self.advance_through(at, self.cursor + 1, portfolio, data, has_pending_orders)?;
+        Ok(applications.pop())
+    }
+
+    fn advance_through(
+        &mut self, at: DateTime<Utc>, end: usize, portfolio: &mut PortfolioState,
+        data: &DataSet, has_pending_orders: bool,
     ) -> Result<Vec<ManualReplayApplication>, String> {
         if at > self.replay.observation_cutoff {
             return Err("manual observation clock exceeds the frozen evidence cutoff".into());
@@ -492,13 +517,6 @@ impl ManualReplayCursor {
         if self.clock.is_some_and(|clock| at < clock) {
             return Err("manual observation clock moved backwards".into());
         }
-        let end = self.cursor
-            + self.indices[self.cursor..]
-                .iter()
-                .take_while(|&&(a, o, f)| {
-                    self.replay.actions[a].orders[o].fills[f].observed_at <= at
-                })
-                .count();
         if end == self.cursor {
             self.clock = Some(at);
             return Ok(vec![]);