fix(bridge): 事件测试钩子在读到 events.dispatch 时计数，修复 order-across-fast-path 偶发失败

crates/rutis-bridge/src/events/hub.rs

@@ -431,6 +431,24 @@ impl SessionEvents {
         }
     }
 
+    /// The `events.*` operation `method` was read, on the session's reader
+    /// (before its next frame; [`SessionEvents::invoke`] runs it later, in
+    /// its turn): the tap counts the dispatches here.
+    pub(crate) fn arrived(&self, method: &str) {
+        #[cfg(feature = "testing")]
+        if method == "events.dispatch" {
+            if let Some(tap) = self
+                .hub
+                .as_ref()
+                .and_then(|hub| hub.tap.lock().unwrap().clone())
+            {
+                tap.received();
+            }
+        }
+        #[cfg(not(feature = "testing"))]
+        let _ = method;
+    }
+
     /// The `events.*` operations a runtime sends.
     pub(crate) fn invoke(
         self: &Arc<Self>,
@@ -449,7 +467,7 @@ impl SessionEvents {
                     .as_ref()
                     .and_then(|hub| hub.tap.lock().unwrap().clone())
                 {
-                    tap.arrived();
+                    tap.starting();
                 }
                 self.dispatch(connection, first)
             }

crates/rutis-bridge/src/events/testing.rs

@@ -69,7 +69,9 @@ impl Tap {
         Ok(())
     }
 
-    /// The `events.dispatch` requests received so far.
+    /// The `events.dispatch` requests received so far: each is counted as
+    /// its session reads it, so one a runtime sent before answering a call
+    /// is counted once that answer is.
     pub fn dispatches(&self) -> u64 {
         self.0.dispatches.load(Ordering::SeqCst)
     }
@@ -88,10 +90,16 @@ impl Tap {
 }
 
 impl TapState {
-    /// An `events.dispatch` arrived: counted, and held if a hold waits for
-    /// it (the thread blocks until the hold is complete).
-    pub(crate) fn arrived(&self) {
+    /// An `events.dispatch` was read: counted on the session's reader,
+    /// before its next frame, so a reply the runtime sent after it (the
+    /// emitter's) finds it counted.
+    pub(crate) fn received(&self) {
         self.dispatches.fetch_add(1, Ordering::SeqCst);
+    }
+
+    /// An `events.dispatch` is about to run: held if a hold waits for it
+    /// (the thread blocks until the hold is complete).
+    pub(crate) fn starting(&self) {
         let mut hold = self.hold.lock().unwrap();
         if hold.want == 0 {
             return;

crates/rutis-bridge/src/peer.rs

@@ -26,6 +26,9 @@ use tokio::sync::watch;
 /// Serves one family of operations on a peer's session.
 pub trait Handler: Send + Sync + 'static {
     fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply;
+
+    /// [`Dispatch::arrived`] for this family's event operations.
+    fn arrived(&self, _peer: &Connection, _method: &str) {}
 }
 
 impl<F> Handler for F
@@ -148,6 +151,14 @@ impl Operations {
 }
 
 impl Dispatch for Operations {
+    fn arrived(&self, peer: &Connection, method: &str) {
+        let handler = (self.handlers.lock().unwrap().get(family("", method)))
+            .map(|family| family.handler.clone());
+        if let Some(handler) = handler {
+            handler.arrived(peer, method);
+        }
+    }
+
     fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply {
         if target.is_empty() && method == "link.offers" {
             let announced = args

crates/rutis-bridge/src/runtime/access.rs

@@ -73,6 +73,10 @@ impl crate::Handler for Routed {
     fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply {
         self.0.invoke(peer, target, method, args)
     }
+
+    fn arrived(&self, peer: &Connection, method: &str) {
+        self.0.arrived(peer, method)
+    }
 }
 
 impl RuntimeSession for PeerSession {

crates/rutis-bridge/src/runtime/process.rs

@@ -450,6 +450,12 @@ impl Imports {
     }
 }
 impl Dispatch for Imports {
+    fn arrived(&self, _connection: &Connection, method: &str) {
+        if let Some(events) = self.bus.get() {
+            events.arrived(method);
+        }
+    }
+
     fn invoke(&self, connection: &Connection, target: &str, method: &str, args: RpcValue) -> Reply {
         if let Some(name) = target.strip_prefix("host:") {
             let (host, row) = self

crates/rutis-bridge/src/session/rpc.rs

@@ -891,6 +891,14 @@ impl Drop for Flight {
 
 pub trait Dispatch: Send + Sync + 'static {
     fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply;
+
+    /// The event operation `method` (an `events.*` control call) was read
+    /// and admitted: called on the reader, before the next frame is taken,
+    /// whereas [`Dispatch::invoke`] runs it later, in the session's order
+    /// of event operations. It must not block. The host's bus counts the
+    /// `events.dispatch` requests here for its test hooks (feature
+    /// `testing`); nothing else needs it.
+    fn arrived(&self, _peer: &Connection, _method: &str) {}
 }
 /// How a session identifies itself and its calls.
 #[derive(Clone, Debug, Default)]
@@ -2292,10 +2300,14 @@ impl Connection {
         // table changes and emits each in its turn, the rest started in it.
         // Never here: a reader runs no plugin code (a listener, an
         // observer), which could wait for a frame only this reader reads.
-        let event_op = matches!(
-            &incoming.call,
-            Accepted::Invoke(target, method, _) if target.is_empty() && method.starts_with("events.")
-        );
+        let event_op = match &incoming.call {
+            Accepted::Invoke(target, method, _)
+                if target.is_empty() && method.starts_with("events.") =>
+            {
+                Some(method.clone())
+            }
+            _ => None,
+        };
         let inline = matches!(
             &incoming.call,
             Accepted::Invoke(target, method, args) if target.is_empty() && inline_event_op(method, args)
@@ -2340,7 +2352,12 @@ impl Connection {
                 Err(id)
             }
         };
-        if let (Err(id), true) = (&delivery, event_op) {
+        if let Some(method) = &event_op {
+            // Arrived now, in frame order: before any later frame of the
+            // session (a reply among them) is read.
+            self.0.dispatch.arrived(self, method);
+        }
+        if let (Err(id), true) = (&delivery, event_op.is_some()) {
             self.run_ordered(id.clone(), inline);
         } else if let Err(id) = delivery {
             let peer = self.clone();