node: observe rekey-msg2 and cross-connection resolution in the per-peer machine

Feed the per-peer machine two observation events so its control/rekey
shadow state stays coherent with the inline crypto-session installs,
without moving those installs. The crypto effect bodies are unchanged.

On rekey-msg2 completion, feed the machine the responder index on the
success path only (after the pending session is installed); the abandon
path feeds nothing.

On cross-connection resolution, feed the machine the swap outcome: on a
swap it records the new outbound and responder indices to mirror the
session replacement, and on a keep it does nothing. The observation
targets the promoted peer's machine, not the outbound leg whose machine
is removed as the block is entered.

Both observation handlers emit no action; nothing consumes the updated
shadow state yet.

src/node/handlers/handshake.rs

@@ -7,7 +7,8 @@ use crate::node::dataplane::PeerActionCtx;
 use crate::node::reject::{HandshakeReject, RejectReason};
 use crate::node::{Node, NodeError};
 use crate::peer::machine::{
-    FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, TimerKind,
+    CrossConnOutcome, FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState,
+    TimerKind,
 };
 use crate::peer::{ActivePeer, PeerConnection};
 use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2};
@@ -16,6 +17,7 @@ use crate::proto::fmp::{
     OutboundSnapshot, PromotionResult, WireOutcome, cross_connection_winner,
 };
 use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
+use crate::utils::index::SessionIndex;
 use std::time::Duration;
 use tracing::{debug, info, warn};
 
@@ -62,6 +64,58 @@ impl EstablishView for Node {
 }
 
 impl Node {
+    /// Feed the peer's control machine the completed-rekey observation after the
+    /// inline `complete_rekey_msg2`. The obs records the peer's new session index
+    /// and advances the rekey phase; it emits no action, so a bare `step` keeps
+    /// the machine coherent without an executor pass.
+    fn observe_rekey_msg2(&mut self, node_addr: &NodeAddr, their_index: SessionIndex) {
+        let link = match self.peers.get(node_addr) {
+            Some(peer) => peer.link_id(),
+            None => return,
+        };
+        if let Some(machine) = self.peer_machines.get_mut(&link) {
+            let acts = machine.step(
+                PeerEvent::RekeyMsg2 { their_index },
+                Self::now_ms(),
+                &mut self.index_allocator,
+            );
+            debug_assert!(acts.is_empty(), "completed-rekey is a pure observation");
+        } else {
+            debug_assert!(
+                false,
+                "peer machine present for every established rekey peer"
+            );
+        }
+    }
+
+    /// Feed the promoted peer's control machine the cross-connection resolution
+    /// after the inline session surgery. The obs reconciles the machine's shadow
+    /// session indices (updated on a swap, unchanged on a keep); it emits no
+    /// action, so a bare `step` keeps the machine coherent without an executor
+    /// pass.
+    fn observe_cross_conn_resolved(&mut self, node_addr: &NodeAddr, outcome: CrossConnOutcome) {
+        let link = match self.peers.get(node_addr) {
+            Some(peer) => peer.link_id(),
+            None => return,
+        };
+        if let Some(machine) = self.peer_machines.get_mut(&link) {
+            let acts = machine.step(
+                PeerEvent::CrossConnResolved { outcome },
+                Self::now_ms(),
+                &mut self.index_allocator,
+            );
+            debug_assert!(
+                acts.is_empty(),
+                "cross-connection resolution is a pure observation"
+            );
+        } else {
+            debug_assert!(
+                false,
+                "peer machine present for the promoted cross-connection peer"
+            );
+        }
+    }
+
     /// Returns true if an inbound msg1 should be admitted past the
     /// `accept_connections` gate.
     ///
@@ -853,6 +907,7 @@ impl Node {
                 let display_name = self.peer_display_name(&peer_node_addr);
 
                 // Complete the rekey handshake on the ActivePeer
+                let mut rekey_completed = false;
                 if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
                     match peer.complete_rekey_msg2(noise_msg2) {
                         Ok((session, remote_epoch)) => {
@@ -890,6 +945,7 @@ impl Node {
                                 new_their_index = %header.sender_idx,
                                 "Rekey completed (initiator), pending K-bit cutover"
                             );
+                            rekey_completed = true;
                         }
                         Err(e) => {
                             warn!(
@@ -909,6 +965,15 @@ impl Node {
                     }
                 }
 
+                // Feed the control machine the completed-rekey observation so its
+                // shadow index and rekey phase stay coherent. Only on success —
+                // the failure path above reverts the rekey and leaves the machine
+                // untouched. The crypto effect already ran inline; this emits no
+                // action.
+                if rekey_completed {
+                    self.observe_rekey_msg2(&peer_node_addr, header.sender_idx);
+                }
+
                 self.pending_outbound.remove(&key);
                 return;
             }
@@ -1026,6 +1091,7 @@ impl Node {
                 }
             };
 
+            let mut cross_conn_outcome: Option<CrossConnOutcome> = None;
             if out_decision == OutboundDecision::CrossConnectionSwap {
                 // We're the smaller node. Swap to outbound session + indices.
                 // The peer will keep their inbound session (complement of ours).
@@ -1078,6 +1144,11 @@ impl Node {
                         new_their_index = %header.sender_idx,
                         "Cross-connection: swapped to outbound session (our outbound wins)"
                     );
+
+                    cross_conn_outcome = Some(CrossConnOutcome::Swap {
+                        our_index: outbound_our_index,
+                        their_index: header.sender_idx,
+                    });
                 }
             } else {
                 // We're the larger node. Keep our inbound session (it pairs
@@ -1103,6 +1174,18 @@ impl Node {
                 if let Some(idx) = outbound_our_index {
                     let _ = self.index_allocator.free(idx);
                 }
+
+                cross_conn_outcome = Some(CrossConnOutcome::Keep);
+            }
+
+            // Feed the promoted peer's control machine the cross-connection
+            // resolution so its shadow session indices track the inline session
+            // surgery above (updated on a swap, unchanged on a keep). The
+            // outbound leg's machine was removed on entry, so this targets the
+            // still-live promoted peer's machine. The crypto effect already ran
+            // inline; this emits no action.
+            if let Some(outcome) = cross_conn_outcome {
+                self.observe_cross_conn_resolved(&peer_node_addr, outcome);
             }
 
             // Clean up outbound connection state

src/peer/machine.rs

@@ -189,6 +189,20 @@ pub(crate) enum TimerKind {
     Liveness,
 }
 
+/// Outcome of an outbound cross-connection resolution, observed by the control
+/// machine after the shell has already applied the effect inline.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum CrossConnOutcome {
+    /// The outbound session replaced the existing inbound one: the control
+    /// shadow adopts the new local and remote session indices.
+    Swap {
+        our_index: SessionIndex,
+        their_index: SessionIndex,
+    },
+    /// The existing inbound session was kept: the control shadow is unchanged.
+    Keep,
+}
+
 /// An input to the machine. Cross-registry facts ride in the payload as
 /// plain-data snapshots ([`WireOutcome`]/[`EstablishSnapshot`]/[`OutboundSnapshot`])
 /// built shell-side; `now` is the `step` parameter, never duplicated here.
@@ -248,6 +262,11 @@ pub(crate) enum PeerEvent {
     /// `Maintaining{Rekey(Msg1Sent)}` so the next tick's `Cutover`/`Drain` consume
     /// transitions from a coherent phase. Emits no action.
     RekeyInitiated,
+    /// OBSERVATION: the shell resolved an outbound cross-connection inline (a
+    /// session swap or keep, with the registry and index surgery already
+    /// applied). Reconciles the control shadow's session indices with reality
+    /// on a swap; leaves them untouched on a keep. Emits no action.
+    CrossConnResolved { outcome: CrossConnOutcome },
     /// Data plane observed the responder K-bit flip inline.
     PeerKbitFlip { epoch: [u8; 8] },
     /// A filter announce is due for this peer.
@@ -521,6 +540,7 @@ impl PeerMachine {
             PeerEvent::RekeyMsg2 { their_index } => self.on_rekey_msg2(their_index),
             PeerEvent::RekeyConsume { action } => self.map_rekey_action(action, now),
             PeerEvent::RekeyInitiated => self.on_rekey_initiated(),
+            PeerEvent::CrossConnResolved { outcome } => self.on_cross_conn_resolved(outcome),
             PeerEvent::PeerKbitFlip { .. } => {
                 // Responder cutover is data-plane-owned: the machine only
                 // schedules the drain-window unregister. NO slot mutation.
@@ -918,6 +938,23 @@ impl PeerMachine {
         Vec::new()
     }
 
+    /// Observation: the shell resolved an outbound cross-connection inline. On a
+    /// session swap it adopts the new local and remote session indices into the
+    /// control shadow, mirroring the shell's in-place session replacement; on a
+    /// keep it leaves the shadow untouched. Emits NO action — the crypto effect
+    /// already ran shell-side.
+    fn on_cross_conn_resolved(&mut self, outcome: CrossConnOutcome) -> Vec<PeerAction> {
+        if let CrossConnOutcome::Swap {
+            our_index,
+            their_index,
+        } = outcome
+        {
+            self.conn.set_our_index(our_index);
+            self.conn.set_their_index(their_index);
+        }
+        Vec::new()
+    }
+
     /// Rekey cadence: run `poll_rekey` over this one peer's snapshot and map the
     /// phase-grouped `ConnAction`s.
     fn on_rekey_cadence(&mut self, now: u64) -> Vec<PeerAction> {
@@ -2437,4 +2474,95 @@ mod tests {
         // No index allocation happened in the machine (shell-side leaf).
         assert_eq!(alloc.count(), 0);
     }
+
+    // ---- Test 11: RekeyMsg2 observation -----------------------------------
+    // The shell completed the initiated rekey inline; the obs records the peer's
+    // new index, clears the in-progress flag, advances to PendingCutover, and
+    // emits nothing.
+    #[test]
+    fn rekey_msg2_observation() {
+        let mut alloc = IndexAllocator::new();
+        let id = peer_identity();
+        let addr = *id.node_addr();
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0);
+        m.state = PeerState::Maintaining {
+            addr,
+            kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent),
+        };
+        m.rekey_in_progress = true;
+
+        let their = SessionIndex::new(0x4444);
+        let acts = m.step(
+            PeerEvent::RekeyMsg2 { their_index: their },
+            6_000,
+            &mut alloc,
+        );
+        assert!(acts.is_empty());
+        assert_eq!(m.conn.their_index(), Some(their));
+        assert!(!m.rekey_in_progress);
+        assert_eq!(
+            m.state(),
+            PeerState::Maintaining {
+                addr,
+                kind: MaintainKind::Rekey(RekeyPhase::PendingCutover)
+            }
+        );
+        assert_eq!(alloc.count(), 0);
+    }
+
+    // ---- Test 12: CrossConnResolved observation ---------------------------
+    // A swap adopts the new local and remote indices into the control shadow and
+    // emits nothing; a keep leaves the shadow untouched and emits nothing.
+    #[test]
+    fn cross_conn_resolved_swap_updates_shadow() {
+        let mut alloc = IndexAllocator::new();
+        let id = peer_identity();
+        let addr = *id.node_addr();
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0);
+        m.state = PeerState::Active { addr };
+        m.conn.set_our_index(SessionIndex::new(0x1111));
+        m.conn.set_their_index(SessionIndex::new(0x2222));
+
+        let our = SessionIndex::new(0xAAAA);
+        let their = SessionIndex::new(0xBBBB);
+        let acts = m.step(
+            PeerEvent::CrossConnResolved {
+                outcome: CrossConnOutcome::Swap {
+                    our_index: our,
+                    their_index: their,
+                },
+            },
+            7_000,
+            &mut alloc,
+        );
+        assert!(acts.is_empty());
+        assert_eq!(m.conn.our_index(), Some(our));
+        assert_eq!(m.conn.their_index(), Some(their));
+        assert_eq!(m.state(), PeerState::Active { addr });
+        assert_eq!(alloc.count(), 0);
+    }
+
+    #[test]
+    fn cross_conn_resolved_keep_is_noop() {
+        let mut alloc = IndexAllocator::new();
+        let id = peer_identity();
+        let addr = *id.node_addr();
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0);
+        m.state = PeerState::Active { addr };
+        m.conn.set_our_index(SessionIndex::new(0x1111));
+        m.conn.set_their_index(SessionIndex::new(0x2222));
+
+        let acts = m.step(
+            PeerEvent::CrossConnResolved {
+                outcome: CrossConnOutcome::Keep,
+            },
+            7_000,
+            &mut alloc,
+        );
+        assert!(acts.is_empty());
+        assert_eq!(m.conn.our_index(), Some(SessionIndex::new(0x1111)));
+        assert_eq!(m.conn.their_index(), Some(SessionIndex::new(0x2222)));
+        assert_eq!(m.state(), PeerState::Active { addr });
+        assert_eq!(alloc.count(), 0);
+    }
 }