Declare a rekey explicitly in msg3 instead of inferring it from session age

A rekey and a fresh dial both arrive as a new msg1 on a new link, so the
responder has to tell them apart at msg3. It cannot: the distinguishing
fact is internal to the initiator and nothing on the wire carries it.
Session age stood in for it and could not do the job - a
message-count-triggered rekey fires on a young session, so a real rekey
landed below the floor, was resolved as a cross-connection, and left the
two ends of the link on different session indices.

Carry the answer instead. The initiator adds a negotiation TLV to its
rekey msg3 naming the session it replaces, using the index the receiver
allocated so the receiver can match it against its own. The shell
resolves that into a three-valued claim - none, matching, mismatched -
and the classifier branches on it.

A mismatched marker resends msg2 rather than rejecting: the initiator has
already installed its pending session and will cut over on its own timer,
and the reject path sends nothing back, so tearing down would strand the
link in the way this defect already does. A fresh dial arriving while our
own rekey is in flight also resends, because the cross-connection swap
rewrites both indices and touches nothing else, which would leave that
rekey pointing at a session the counterpart no longer holds. The age
floor was excluding that case incidentally.

The floor and the session-age snapshot field are gone with it, so the
establish snapshot no longer reads the clock at all.

Both node-level rekey tests now drive a real rekey through the
initiator's own trigger. They previously reached the rekey arms by
backdating a session and re-handshaking, which tested the age proxy this
replaces; one of them was reaching its assertions through the duplicate
arm rather than the arm it names. Age them well past the trigger, since
the jitter is +/-15s and a smaller margin makes firing depend on the draw.

Incomplete: no TLV round-trip test, no two-ended convergence test, no
rekey-crossed-with-fresh-dial case, and no integration run yet.

src/node/handlers/handshake.rs

@@ -19,8 +19,8 @@ use crate::proto::fmp::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, bu
 use crate::proto::fmp::{
     DialMsg2Decision, DialMsg2Reject, DialMsg2Snapshot, Disconnect, DisconnectReason,
     EstablishSnapshot, InboundDecision, InboundReject, NegotiationPayload, OutboundSnapshot,
-    PromotionResult, RekeyMsg2Decision, RekeyMsg2Reject, RekeyMsg2Snapshot, WireOutcome,
-    cross_connection_winner, decide_fmp_negotiation,
+    PromotionResult, RekeyClaim, RekeyMsg2Decision, RekeyMsg2Reject, RekeyMsg2Snapshot,
+    WireOutcome, cross_connection_winner, decide_fmp_negotiation,
 };
 use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
 use crate::utils::index::SessionIndex;
@@ -478,8 +478,9 @@ impl Node {
                 let msg3_now_ms = Self::now_ms();
 
                 let mut rekey_completed = false;
+                let our_profile = self.node_profile();
                 if let Some(peer) = self.peers.get_mut(&peer_node_addr) {
-                    match peer.complete_rekey_msg2(noise_msg2) {
+                    match peer.complete_rekey_msg2(noise_msg2, our_profile) {
                         Ok((msg3_bytes, session, remote_epoch, learned_peer)) => {
                             // Static-key continuity gate. The rekey msg2 was
                             // matched to this peer by the session index WE
@@ -1215,7 +1216,7 @@ impl Node {
         };
 
         let our_profile = self.node_profile();
-        let (peer_identity, our_index, remote_epoch) = {
+        let (peer_identity, our_index, remote_epoch, declared_rekey_of) = {
             // Get the pending connection
             let machine = match self.peer_machines.get_mut(&link_id) {
                 Some(m) => m,
@@ -1289,7 +1290,27 @@ impl Node {
             let our_index = machine.our_index();
             let remote_epoch = machine.conn_remote_epoch();
 
-            (peer_identity, our_index, remote_epoch)
+            // The sender's rekey declaration. Decoded again here rather than
+            // threaded out of `process_fmp_negotiation` so the msg2 path is
+            // untouched; the payload is a few bytes. A present-but-malformed
+            // marker is a hard failure: reading it as "not a rekey" would be
+            // exactly the silent fall-through this field exists to remove.
+            let declared_rekey_of = match &received_negotiation {
+                Some(bytes) => match NegotiationPayload::decode(bytes).and_then(|p| p.rekey_of()) {
+                    Ok(claim) => claim,
+                    Err(e) => {
+                        warn!(link_id = %link_id, error = %e, "Malformed rekey marker in msg3");
+                        self.remove_link(&link_id);
+                        self.remove_peer_machine(link_id);
+                        self.stats_mut()
+                            .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
+                        return;
+                    }
+                },
+                None => None,
+            };
+
+            (peer_identity, our_index, remote_epoch, declared_rekey_of)
         };
 
         let peer_node_addr = *peer_identity.node_addr();
@@ -1372,24 +1393,28 @@ impl Node {
         // running the actions through the executor and owning only the residual
         // shell bookkeeping (link/map removal, reject records, the duplicate-msg2
         // resend). The snapshot resolves the one clock read (session age) and the
-        // config-derived rekey floor up front.
+        // sender's rekey declaration up front.
         //
-        // The rekey age floor sits BELOW the minimum possible rekey interval, or
-        // jittered rekeys are wrongly rejected. It bounds both the
-        // cross-connection branch (`< floor` -> initial cross-connection) and the
-        // rekey-responder branch (`>= floor` -> rekey), so the two partition
-        // cleanly; see the pre-refactor commentary retained on the decision.
+        // The declaration replaces a session-age floor that used to partition
+        // "too young to be a rekey" from "old enough to be one". That floor was
+        // never sound: a message-count-triggered rekey fires on a young session,
+        // so a real rekey could land below it and be resolved as a
+        // cross-connection, leaving the two ends on different session indices.
         let our_node_addr = *self.identity().node_addr();
         let rekey_enabled = self.config().node.rekey.enabled;
-        let rekey_age_floor_secs = {
-            let min_interval = self
-                .config()
-                .node
-                .rekey
-                .after_secs
-                .saturating_sub(crate::node::REKEY_JITTER_SECS.unsigned_abs());
-            min_interval.saturating_sub(5).max(5)
+
+        // Resolve the sender's declaration against the session we hold. Matched
+        // ONLY against the peer this handshake authenticated as, never through a
+        // global index map — otherwise one peer could name another's index and
+        // steer a decision about that peer's session.
+        let rekey_claim = match declared_rekey_of {
+            None => RekeyClaim::None,
+            Some(declared) => match self.peers.get(&peer_node_addr).and_then(|p| p.our_index()) {
+                Some(ours) if ours == declared => RekeyClaim::Matches,
+                _ => RekeyClaim::Mismatch,
+            },
         };
+
         let wire = WireOutcome {
             peer_node_addr,
             remote_epoch,
@@ -1398,10 +1423,6 @@ impl Node {
             Some(existing_peer) => EstablishSnapshot {
                 has_existing_peer: true,
                 existing_peer_epoch: existing_peer.remote_epoch(),
-                existing_session_age_secs: existing_peer
-                    .session_established_at()
-                    .elapsed()
-                    .as_secs(),
                 has_session: existing_peer.has_session(),
                 is_healthy: existing_peer.is_healthy(),
                 pending_new_session: existing_peer.pending_new_session().is_some(),
@@ -1409,13 +1430,12 @@ impl Node {
                 existing_msg2: existing_peer.handshake_msg2().map(|m| m.to_vec()),
                 different_link: existing_peer.link_id() != link_id,
                 rekey_enabled,
-                rekey_age_floor_secs,
+                rekey_claim,
                 our_node_addr,
             },
             None => EstablishSnapshot {
                 has_existing_peer: false,
                 existing_peer_epoch: None,
-                existing_session_age_secs: 0,
                 has_session: false,
                 is_healthy: false,
                 pending_new_session: false,
@@ -1423,7 +1443,7 @@ impl Node {
                 existing_msg2: None,
                 different_link: false,
                 rekey_enabled,
-                rekey_age_floor_secs,
+                rekey_claim,
                 our_node_addr,
             },
         };

src/node/tests/handshake.rs

@@ -1582,10 +1582,34 @@ async fn drive_to_msg3(
     recv_phase(&mut responder.packet_rx, 3, "msg3").await
 }
 
+/// Drive a **real** rekey: the initiator's own `check_rekey` builds the rekey
+/// msg1, and the msg3 that returns carries the rekey marker.
+///
+/// A bare second handshake will not substitute. It declares no rekey — because
+/// it is not one — so the responder reads it as a fresh dial. Reaching the rekey
+/// arm by backdating a session and re-handshaking tested the age proxy that the
+/// marker replaces, not the rekey path.
+///
+/// Both ends' sessions must be aged past the trigger before calling this. Age
+/// them well past it: the trigger is `after_secs + jitter` with jitter drawn from
+/// [-15, +15], so a margin under 15s makes firing depend on the draw and the test
+/// flaky.
+async fn drive_rekey_to_msg3(initiator: &mut HsNode, responder: &mut HsNode) -> ReceivedPacket {
+    initiator.node.check_rekey().await;
+
+    let msg1_pkt = recv_phase(&mut responder.packet_rx, 1, "rekey msg1").await;
+    responder.node.handle_msg1(msg1_pkt).await;
+
+    let msg2_pkt = recv_phase(&mut initiator.packet_rx, 2, "rekey msg2").await;
+    initiator.node.handle_msg2(msg2_pkt).await;
+
+    recv_phase(&mut responder.packet_rx, 3, "rekey msg3").await
+}
+
 #[tokio::test]
 async fn test_msg3_dual_rekey_won_frees_index() {
-    // Rekey enabled with a tiny interval so the rekey age floor collapses to
-    // its 5s minimum; the peer session is then backdated past it.
+    // Both ends carry the rekey config: the initiator to fire its own trigger,
+    // the responder for the `rekey_enabled` gate in the classifier.
     let make_config = || {
         let mut c = Config::new();
         c.node.rekey.enabled = true;
@@ -1593,7 +1617,7 @@ async fn test_msg3_dual_rekey_won_frees_index() {
         c
     };
 
-    let mut initiator = make_hs_node(Config::new()).await;
+    let mut initiator = make_hs_node(make_config()).await;
     // The DualRekeyWon tie-break is won by the numerically smaller node addr,
     // so the responder (whose handle_msg3 we exercise) must be the smaller.
     let mut responder = loop {
@@ -1611,19 +1635,28 @@ async fn test_msg3_dual_rekey_won_frees_index() {
     let baseline = responder.node.index_allocator.count();
     assert_eq!(baseline, 1, "responder holds exactly the peer's index");
 
-    // Age the session past the rekey floor and mark a rekey in progress so a
-    // fresh inbound msg3 classifies as the dual-init rekey we win.
+    // Mark a rekey of our own in progress so the peer's declared rekey meets it
+    // as a dual initiation, which the smaller node addr wins. Age both ends so
+    // the initiator's own trigger fires and produces a real, marked rekey msg3 —
+    // a bare handshake declares no rekey and would never reach this arm.
     let peer_addr =
         *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr();
+    let responder_addr =
+        *PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full()).node_addr();
+    initiator
+        .node
+        .get_peer_mut(&responder_addr)
+        .unwrap()
+        .test_backdate_session_established(std::time::Duration::from_secs(120));
     {
         let peer = responder.node.get_peer_mut(&peer_addr).unwrap();
-        peer.test_backdate_session_established(std::time::Duration::from_secs(6));
+        peer.test_backdate_session_established(std::time::Duration::from_secs(120));
         peer.set_rekey_in_progress();
     }
 
-    // Second handshake: the new inbound msg1 allocates a fresh index, then the
-    // msg3 lands on the DualRekeyWon reject arm.
-    let msg3b = drive_to_msg3(&mut initiator, &mut responder, 2000).await;
+    // The rekey's msg1 allocates a fresh index on the responder, then its msg3
+    // lands on the DualRekeyWon reject arm.
+    let msg3b = drive_rekey_to_msg3(&mut initiator, &mut responder).await;
     assert_eq!(
         responder.node.index_allocator.count(),
         baseline + 1,
@@ -1686,7 +1719,7 @@ async fn test_msg3_resend_msg2_frees_index() {
         .node
         .get_peer_mut(&peer_addr)
         .unwrap()
-        .test_backdate_session_established(std::time::Duration::from_secs(6));
+        .test_backdate_session_established(std::time::Duration::from_secs(120));
 
     // Second (duplicate) handshake: fresh index allocated at msg1, then freed
     // on the ResendMsg2 arm.
@@ -1842,14 +1875,17 @@ async fn test_msg3_crypto_fail_disposes_leg_machine() {
 
 #[tokio::test]
 async fn test_msg3_rekey_respond_disposes_leg_machine() {
-    // Rekey enabled with a tiny interval so the rekey age floor collapses to
-    // its 5s minimum; with the session backdated past it and NO rekey of our
-    // own in flight, a fresh inbound msg3 classifies as rekey-responder.
+    // A REAL rekey, driven from the initiator's own trigger so its msg3 carries
+    // the rekey marker. Both ends need the rekey config: the initiator to fire
+    // the trigger, the responder for the `rekey_enabled` gate in the classifier.
+    let mut init_config = Config::new();
+    init_config.node.rekey.enabled = true;
+    init_config.node.rekey.after_secs = 1;
     let mut config = Config::new();
     config.node.rekey.enabled = true;
     config.node.rekey.after_secs = 1;
 
-    let mut initiator = make_hs_node(Config::new()).await;
+    let mut initiator = make_hs_node(init_config).await;
     let mut responder = make_hs_node(config).await;
 
     // First handshake establishes the active peer (and its machine).
@@ -1859,16 +1895,23 @@ async fn test_msg3_rekey_respond_disposes_leg_machine() {
 
     let peer_addr =
         *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr();
-    responder
-        .node
-        .get_peer_mut(&peer_addr)
-        .unwrap()
-        .test_backdate_session_established(std::time::Duration::from_secs(6));
+    let responder_addr =
+        *PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full()).node_addr();
+    // Age BOTH ends: the initiator so its rekey trigger fires, the responder so
+    // its own state matches what the rekey replaces.
+    for (node, addr) in [
+        (&mut initiator.node, responder_addr),
+        (&mut responder.node, peer_addr),
+    ] {
+        node.get_peer_mut(&addr)
+            .unwrap()
+            .test_backdate_session_established(std::time::Duration::from_secs(120));
+    }
 
-    // Second handshake lands on the rekey-responder arm: the pending session
+    // The rekey's msg3 lands on the rekey-responder arm: the pending session
     // moves onto the established peer; the window leg and its msg1-born
     // machine are consumed.
-    let msg3b = drive_to_msg3(&mut initiator, &mut responder, 2000).await;
+    let msg3b = drive_rekey_to_msg3(&mut initiator, &mut responder).await;
     responder.node.handle_msg3(msg3b).await;
 
     let peer = responder.node.get_peer(&peer_addr).unwrap();

src/peer/active.rs

@@ -1289,6 +1289,7 @@ impl ActivePeer {
     pub fn complete_rekey_msg2(
         &mut self,
         msg2_bytes: &[u8],
+        our_profile: NodeProfile,
     ) -> Result<RekeyMsg2Completion, NoiseError> {
         let mut hs = self
             .rekey_handshake
@@ -1318,7 +1319,21 @@ impl ActivePeer {
             let _ = hs.decrypt_payload(encrypted_neg)?;
         }
 
-        let msg3 = hs.write_message_3()?;
+        // Declare this handshake a rekey of the session already installed, naming
+        // the index the RESPONDER receives on (our `their_index`) so it can match
+        // the marker against its own `our_index`. Without it the responder cannot
+        // tell a rekey from a fresh dial — both arrive as a new msg1 on a new
+        // link — and every attempt to infer that from local state has a failure
+        // band. The pre-rekey session is still installed at this point; the
+        // pending one is not set until the caller drives it.
+        let mut msg3 = hs.write_message_3()?;
+        if let Some(their_index) = self.their_index() {
+            let marker = NegotiationPayload::fmp(1, 1, our_profile)
+                .with_rekey_of(their_index)
+                .encode();
+            let encrypted = hs.encrypt_payload(&marker)?;
+            msg3.extend_from_slice(&encrypted);
+        }
         let session = hs.into_session()?;
 
         // Derive the learned identity from the session rather than the consumed

src/peer/machine.rs

@@ -2230,7 +2230,7 @@ fn disconnect_frame(reason: CloseReason) -> Vec<u8> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::proto::fmp::PromotionResult;
+    use crate::proto::fmp::{PromotionResult, RekeyClaim};
     use crate::{Identity, PeerIdentity};
 
     fn peer_identity() -> PeerIdentity {
@@ -2265,7 +2265,6 @@ mod tests {
         EstablishSnapshot {
             has_existing_peer: false,
             existing_peer_epoch: None,
-            existing_session_age_secs: 0,
             has_session: false,
             is_healthy: false,
             pending_new_session: false,
@@ -2273,7 +2272,7 @@ mod tests {
             existing_msg2: None,
             different_link: false,
             rekey_enabled: true,
-            rekey_age_floor_secs: 60,
+            rekey_claim: RekeyClaim::None,
             our_node_addr: our,
         }
     }
@@ -2700,7 +2699,7 @@ mod tests {
             est.existing_peer_epoch = Some([1u8; 8]);
             est.has_session = true;
             est.is_healthy = true;
-            est.existing_session_age_secs = 120; // >= floor -> rekey path
+            est.rekey_claim = RekeyClaim::Matches; // declared rekey of our session
             est.rekey_in_progress = true;
             let wire = wire_outcome(peer_addr, Some([1u8; 8]));
 
@@ -2747,7 +2746,7 @@ mod tests {
             est.existing_peer_epoch = Some([1u8; 8]);
             est.has_session = true;
             est.is_healthy = true;
-            est.existing_session_age_secs = 120;
+            est.rekey_claim = RekeyClaim::Matches;
             est.rekey_in_progress = true;
             let wire = wire_outcome(peer_addr, Some([1u8; 8]));
 
@@ -3069,7 +3068,7 @@ mod tests {
         est.has_session = true;
         est.is_healthy = true;
         est.different_link = true;
-        est.existing_session_age_secs = 10; // < floor(60) -> cross-connection
+        est.rekey_claim = RekeyClaim::None; // no rekey declared -> cross-connection
         let wire = wire_outcome(peer_addr, Some([5u8; 8]));
 
         let actions = m.step(
@@ -3116,7 +3115,7 @@ mod tests {
         est.has_session = true;
         est.is_healthy = true;
         est.different_link = true;
-        est.existing_session_age_secs = 10; // < floor(60) -> cross-connection
+        est.rekey_claim = RekeyClaim::None; // no rekey declared -> cross-connection
         let wire = wire_outcome(peer_addr, Some([5u8; 8]));
 
         let actions = m.step(
@@ -3161,7 +3160,7 @@ mod tests {
         est.existing_peer_epoch = Some([7u8; 8]);
         est.has_session = true;
         est.is_healthy = true;
-        est.existing_session_age_secs = 120; // >= floor -> rekey path
+        est.rekey_claim = RekeyClaim::Matches; // declared rekey of our session
         let wire = wire_outcome(peer_addr, Some([7u8; 8]));
 
         let actions = m.step(

src/proto/fmp/core.rs

@@ -231,10 +231,6 @@ pub(crate) struct EstablishSnapshot {
     pub has_existing_peer: bool,
     /// The existing active peer's captured remote startup epoch, if any.
     pub existing_peer_epoch: Option<[u8; 8]>,
-    /// Monotonic age in seconds of the existing peer's session
-    /// (`session_established_at().elapsed()`), resolved shell-side. `0` when
-    /// there is no existing peer.
-    pub existing_session_age_secs: u64,
     /// The existing peer has an established Noise session.
     pub has_session: bool,
     /// The existing peer's session is healthy.
@@ -255,19 +251,35 @@ pub(crate) struct EstablishSnapshot {
     /// cross-connection branch: a `msg3` on the same link is never a
     /// cross-connection.
     pub different_link: bool,
-    /// The local rekey trigger is enabled in config (gates treating an aged
-    /// same-epoch `msg3` as a rekey rather than a duplicate).
+    /// The local rekey trigger is enabled in config (gates treating a declared
+    /// rekey as a rekey rather than a duplicate).
     pub rekey_enabled: bool,
-    /// The config-derived minimum session age (seconds) that partitions an aged
-    /// rekey from a fresh cross-connection: `< floor` → initial
-    /// cross-connection, `>= floor` → rekey responder. Derived shell-side from
-    /// `rekey.after_secs` and the rekey jitter so it tracks the real minimum
-    /// rekey spacing.
-    pub rekey_age_floor_secs: u64,
+    /// What the sender's `msg3` declares about replacing an existing session,
+    /// resolved shell-side against the peer at `wire.peer_node_addr`.
+    pub rekey_claim: RekeyClaim,
     /// This node's own address, for both tie-breaks (the smaller NodeAddr wins).
     pub our_node_addr: NodeAddr,
 }
 
+/// What an inbound `msg3` declares about replacing a session we already hold.
+///
+/// The sender knows whether it is rekeying; the responder cannot infer it, since
+/// a rekey and a fresh dial both arrive as a new `msg1` on a new link. The msg3
+/// negotiation TLV carries the sender's answer and the shell resolves it into
+/// this three-valued input — resolved shell-side because matching the declared
+/// index against the session we hold is a registry read, and because it must be
+/// matched only against the peer the handshake authenticated as, never resolved
+/// through a global index map.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum RekeyClaim {
+    /// No marker: the sender says this handshake is not a rekey.
+    None,
+    /// The marker names the session we hold for this peer.
+    Matches,
+    /// The marker names a session that is not the one we hold.
+    Mismatch,
+}
+
 /// A snapshot of the registry state the *outbound* establish decision reads
 /// about the peer whose msg2 just completed our handshake, taken by the shell.
 ///
@@ -731,11 +743,41 @@ impl Fmp {
             _ => {
                 // Same epoch (or no epoch captured on either side).
 
-                // Inline cross-connection (msg2-then-msg3 ordering): a
-                // still-fresh session receiving a concurrent msg3 on a different
-                // link. The upper age bound sits below the rekey floor so any
-                // rekey-aged msg3 falls through to the rekey responder path.
-                if snap.different_link && snap.existing_session_age_secs < snap.rekey_age_floor_secs
+                // A rekey and a fresh dial both arrive as a new msg1 on a new
+                // link, so the sender declares which it is: the msg3 negotiation
+                // TLV names the session it replaces and the shell resolves that
+                // against the session we hold. Session age used to stand in for
+                // this and could not — a message-count-triggered rekey fires on a
+                // young session and was misread as a cross-connection, which is
+                // the defect this replaces.
+                if snap.rekey_claim == RekeyClaim::Mismatch {
+                    // The sender is replacing a session that is not the one we
+                    // hold. Resend rather than reject: the initiator installed its
+                    // pending session at msg2 completion and will cut over on its
+                    // own timer, and the reject path sends nothing back, so
+                    // tearing down here strands the link in exactly the way this
+                    // defect does. Resending leaves both current sessions intact
+                    // and lets the initiator's resend/abandon machinery converge.
+                    return InboundDecision::ResendMsg2 {
+                        msg2: snap.existing_msg2.clone(),
+                    };
+                }
+
+                // The sender declares no rekey, so a msg3 on a different link is
+                // a fresh dial crossing ours.
+                //
+                // A rekey of ours in flight suppresses the swap. `replace_session`
+                // rewrites the session and both indices and touches nothing else,
+                // so swapping while a pending rekey exists leaves that rekey and
+                // its K-bit state pointing at a session the counterpart no longer
+                // holds. The age floor excluded this case incidentally; the marker
+                // does not, so it is stated.
+                if snap.rekey_claim == RekeyClaim::None
+                    && snap.different_link
+                    && snap.has_session
+                    && snap.is_healthy
+                    && !snap.rekey_in_progress
+                    && !snap.pending_new_session
                 {
                     // `cross_connection_winner(our, peer, this_is_outbound=false)`:
                     // the smaller node prefers its outbound, so our *inbound*
@@ -748,11 +790,11 @@ impl Fmp {
                     };
                 }
 
-                // Rekey responder gate: aged, healthy session with rekey enabled.
-                if snap.rekey_enabled
+                // Rekey responder gate: the sender named the session we hold.
+                if snap.rekey_claim == RekeyClaim::Matches
+                    && snap.rekey_enabled
                     && snap.has_session
                     && snap.is_healthy
-                    && snap.existing_session_age_secs >= snap.rekey_age_floor_secs
                 {
                     // Widened dual-init tie-break: both the still-in-progress and
                     // the already-pending states resolve by the smaller NodeAddr.

src/proto/fmp/mod.rs

@@ -43,7 +43,7 @@ mod tests;
 pub(crate) use core::{
     ConnAction, ConnSnapshot, DialMsg2Decision, DialMsg2Reject, DialMsg2Snapshot,
     EstablishSnapshot, InboundDecision, InboundReject, LifecycleView, OutboundDecision,
-    OutboundSnapshot, PeerSnapshot, RekeyCfg, RekeyMsg2Decision, RekeyMsg2Reject,
+    OutboundSnapshot, PeerSnapshot, RekeyCfg, RekeyClaim, RekeyMsg2Decision, RekeyMsg2Reject,
     RekeyMsg2Snapshot, RekeyResendSnapshot, WireOutcome, decide_fmp_negotiation,
 };
 pub use core::{PromotionResult, cross_connection_winner};

src/proto/fmp/tests/core.rs

@@ -7,7 +7,7 @@ use super::util::{
 use crate::proto::fmp::{
     ConnAction, DialMsg2Decision, DialMsg2Reject, DialMsg2Snapshot, Fmp, InboundDecision,
     InboundReject, NegotiationPayload, NodeProfile, OutboundDecision, OutboundSnapshot, RekeyCfg,
-    RekeyMsg2Decision, RekeyMsg2Reject, RekeyMsg2Snapshot, cross_connection_winner,
+    RekeyClaim, RekeyMsg2Decision, RekeyMsg2Reject, RekeyMsg2Snapshot, cross_connection_winner,
 };
 use crate::testutil::make_node_addr;
 use crate::transport::LinkId;
@@ -352,7 +352,7 @@ fn establish_cross_connection_larger_node_inbound_wins() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x09); // our addr 0x09
     snap.different_link = true;
-    snap.existing_session_age_secs = 0; // < floor
+    snap.rekey_claim = RekeyClaim::None; // sender declares no rekey
     let wire = wire_outcome(0x02, SAME_EPOCH); // peer 0x02 < our 0x09
     match fmp.establish_inbound(&snap, &wire) {
         InboundDecision::CrossConnect {
@@ -373,7 +373,7 @@ fn establish_cross_connection_smaller_node_inbound_loses() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x02); // our addr 0x02
     snap.different_link = true;
-    snap.existing_session_age_secs = 0;
+    snap.rekey_claim = RekeyClaim::None;
     let wire = wire_outcome(0x09, SAME_EPOCH); // peer 0x09 > our 0x02
     match fmp.establish_inbound(&snap, &wire) {
         InboundDecision::CrossConnect {
@@ -390,7 +390,7 @@ fn establish_same_link_fresh_is_duplicate() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x05);
     snap.different_link = false;
-    snap.existing_session_age_secs = 0;
+    snap.rekey_claim = RekeyClaim::None;
     snap.existing_msg2 = Some(vec![0xaa, 0xbb]);
     let wire = wire_outcome(0x02, SAME_EPOCH);
     match fmp.establish_inbound(&snap, &wire) {
@@ -406,7 +406,7 @@ fn establish_aged_rekey_disabled_is_duplicate() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x05);
     snap.rekey_enabled = false;
-    snap.existing_session_age_secs = 300; // >= floor, but rekey disabled
+    snap.rekey_claim = RekeyClaim::Matches; // declared, but rekey disabled
     let wire = wire_outcome(0x02, SAME_EPOCH);
     assert!(matches!(
         fmp.establish_inbound(&snap, &wire),
@@ -420,7 +420,7 @@ fn establish_aged_rekey_disabled_is_duplicate() {
 fn establish_aged_rekey_responds_without_abandon() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x05);
-    snap.existing_session_age_secs = 300; // >= floor
+    snap.rekey_claim = RekeyClaim::Matches;
     let wire = wire_outcome(0x02, SAME_EPOCH);
     match fmp.establish_inbound(&snap, &wire) {
         InboundDecision::RekeyRespond {
@@ -440,7 +440,7 @@ fn establish_aged_rekey_responds_without_abandon() {
 fn establish_dual_init_in_progress_smaller_wins() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x02); // our addr 0x02 (smaller)
-    snap.existing_session_age_secs = 300;
+    snap.rekey_claim = RekeyClaim::Matches;
     snap.rekey_in_progress = true;
     let wire = wire_outcome(0x09, SAME_EPOCH); // peer 0x09
     assert!(matches!(
@@ -457,7 +457,7 @@ fn establish_dual_init_in_progress_smaller_wins() {
 fn establish_dual_init_in_progress_larger_loses() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x09); // our addr 0x09 (larger)
-    snap.existing_session_age_secs = 300;
+    snap.rekey_claim = RekeyClaim::Matches;
     snap.rekey_in_progress = true;
     let wire = wire_outcome(0x02, SAME_EPOCH); // peer 0x02
     match fmp.establish_inbound(&snap, &wire) {
@@ -473,7 +473,7 @@ fn establish_dual_init_in_progress_larger_loses() {
 fn establish_dual_init_pending_state_larger_loses() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x09); // our addr 0x09 (larger)
-    snap.existing_session_age_secs = 300;
+    snap.rekey_claim = RekeyClaim::Matches;
     snap.rekey_in_progress = false;
     snap.pending_new_session = true; // the widened window
     let wire = wire_outcome(0x02, SAME_EPOCH);
@@ -489,7 +489,7 @@ fn establish_dual_init_pending_state_larger_loses() {
 fn establish_dual_init_pending_state_smaller_wins() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x02); // our addr 0x02 (smaller)
-    snap.existing_session_age_secs = 300;
+    snap.rekey_claim = RekeyClaim::Matches;
     snap.pending_new_session = true;
     let wire = wire_outcome(0x09, SAME_EPOCH);
     assert!(matches!(
@@ -500,38 +500,83 @@ fn establish_dual_init_pending_state_smaller_wins() {
     ));
 }
 
-/// The rekey floor partitions cross-connection from rekey exactly at the
-/// boundary: one second below is a cross-connection, at the floor is a rekey.
+/// The sender's declaration partitions cross-connection from rekey, and it does
+/// so independently of session age. This is the property the age floor could not
+/// provide: a message-count-triggered rekey fires on a young session, so an
+/// age-based partition classified a real rekey as a cross-connection and left the
+/// two ends on different session indices.
 #[test]
-fn establish_rekey_floor_partitions_cross_connection_and_rekey() {
+fn establish_declaration_partitions_cross_connection_and_rekey() {
     let fmp = Fmp::new();
     let wire = wire_outcome(0x02, SAME_EPOCH);
 
-    let mut below = establish_snapshot(0x09);
-    below.different_link = true;
-    below.rekey_age_floor_secs = 100;
-    below.existing_session_age_secs = 99;
+    let mut undeclared = establish_snapshot(0x09);
+    undeclared.different_link = true;
+    undeclared.rekey_claim = RekeyClaim::None;
     assert!(matches!(
-        fmp.establish_inbound(&below, &wire),
+        fmp.establish_inbound(&undeclared, &wire),
         InboundDecision::CrossConnect { .. }
     ));
 
-    let mut at = establish_snapshot(0x09);
-    at.different_link = true;
-    at.rekey_age_floor_secs = 100;
-    at.existing_session_age_secs = 100;
+    let mut declared = establish_snapshot(0x09);
+    declared.different_link = true;
+    declared.rekey_claim = RekeyClaim::Matches;
     assert!(matches!(
-        fmp.establish_inbound(&at, &wire),
+        fmp.establish_inbound(&declared, &wire),
         InboundDecision::RekeyRespond { .. }
     ));
 }
 
+/// A marker naming a session we do not hold resends msg2 rather than rejecting.
+/// The initiator has already installed its pending session and will cut over on
+/// its own timer; the reject path sends nothing back, so tearing down here would
+/// strand the link in exactly the way this defect does.
+#[test]
+fn establish_mismatched_marker_resends_rather_than_rejecting() {
+    let fmp = Fmp::new();
+    let wire = wire_outcome(0x02, SAME_EPOCH);
+
+    let mut snap = establish_snapshot(0x09);
+    snap.different_link = true;
+    snap.rekey_claim = RekeyClaim::Mismatch;
+    assert!(matches!(
+        fmp.establish_inbound(&snap, &wire),
+        InboundDecision::ResendMsg2 { .. }
+    ));
+}
+
+/// A fresh dial arriving while our own rekey is in flight must not swap. The
+/// swap rewrites the session and both indices and touches nothing else, so it
+/// would leave the pending rekey and its K-bit state pointing at a session the
+/// counterpart no longer holds. The age floor excluded this case incidentally;
+/// the declaration does not, so the guard is explicit.
+#[test]
+fn establish_undeclared_during_our_rekey_does_not_swap() {
+    let fmp = Fmp::new();
+    let wire = wire_outcome(0x02, SAME_EPOCH);
+
+    for (in_progress, pending) in [(true, false), (false, true)] {
+        let mut snap = establish_snapshot(0x09);
+        snap.different_link = true;
+        snap.rekey_claim = RekeyClaim::None;
+        snap.rekey_in_progress = in_progress;
+        snap.pending_new_session = pending;
+        assert!(
+            matches!(
+                fmp.establish_inbound(&snap, &wire),
+                InboundDecision::ResendMsg2 { .. }
+            ),
+            "in_progress={in_progress} pending={pending}"
+        );
+    }
+}
+
 /// An unhealthy or session-less aged peer is not a rekey candidate -> duplicate.
 #[test]
 fn establish_aged_unhealthy_is_duplicate() {
     let fmp = Fmp::new();
     let mut snap = establish_snapshot(0x05);
-    snap.existing_session_age_secs = 300;
+    snap.rekey_claim = RekeyClaim::Matches;
     snap.is_healthy = false;
     let wire = wire_outcome(0x02, SAME_EPOCH);
     assert!(matches!(

src/proto/fmp/tests/util.rs

@@ -1,7 +1,7 @@
 //! Shared test helpers for the FMP connection-lifecycle unit tests.
 
 use crate::proto::fmp::{
-    ConnSnapshot, EstablishSnapshot, PeerSnapshot, RekeyResendSnapshot, WireOutcome,
+    ConnSnapshot, EstablishSnapshot, PeerSnapshot, RekeyClaim, RekeyResendSnapshot, WireOutcome,
 };
 use crate::testutil::make_node_addr;
 use crate::transport::LinkId;
@@ -72,14 +72,14 @@ pub(super) fn resend_snapshot(link: LinkId, resend_count: u32, msg1: Vec<u8>) ->
 
 /// Build an `EstablishSnapshot` describing an existing, healthy, same-epoch peer
 /// with a rekey-enabled config and a rekey age floor of 100s, owned by node
-/// `our_byte`. The default is a quiescent, still-fresh session (age 0, same
-/// link, no in-flight rekey / pending). Tests override only the fields their
-/// branch exercises. `existing_peer_epoch` defaults to `[0x01; 8]`.
+/// `our_byte`. The default is a quiescent session on the same link with no
+/// in-flight rekey / pending and no rekey declared by the sender. Tests override
+/// only the fields their branch exercises. `existing_peer_epoch` defaults to
+/// `[0x01; 8]`.
 pub(super) fn establish_snapshot(our_byte: u8) -> EstablishSnapshot {
     EstablishSnapshot {
         has_existing_peer: true,
         existing_peer_epoch: Some([0x01; 8]),
-        existing_session_age_secs: 0,
         has_session: true,
         is_healthy: true,
         pending_new_session: false,
@@ -87,7 +87,7 @@ pub(super) fn establish_snapshot(our_byte: u8) -> EstablishSnapshot {
         existing_msg2: None,
         different_link: false,
         rekey_enabled: true,
-        rekey_age_floor_secs: 100,
+        rekey_claim: RekeyClaim::None,
         our_node_addr: make_node_addr(our_byte),
     }
 }

src/proto/fmp/wire.rs

@@ -193,6 +193,12 @@ pub const NEGOTIATION_HEADER_SIZE: usize = 10;
 /// Format byte value for the initial negotiation format.
 pub(crate) const NEGOTIATION_FORMAT_V0: u8 = 0;
 
+/// TLV field number for the rekey marker.
+///
+/// Its value is the session index the *receiver* allocated for the session this
+/// handshake replaces. Absence means the handshake is not a rekey.
+pub const TLV_REKEY_OF: u16 = 1;
+
 // --- FMP feature bitfield constants ---
 
 /// Mask for the 3-bit node profile enum (bits 0-2).
@@ -296,6 +302,38 @@ impl NegotiationPayload {
         self
     }
 
+    /// Declare this handshake a rekey of the session the receiver indexes as
+    /// `their_index`.
+    ///
+    /// The marker names the *receiver's* index, not the sender's, because the
+    /// receiver matches it against its own `our_index`. Absence means the
+    /// handshake is not a rekey — a responder cannot otherwise tell a rekey from
+    /// a fresh dial, since both arrive as a new msg1 on a new link.
+    pub fn with_rekey_of(self, their_index: SessionIndex) -> Self {
+        self.with_tlv(TLV_REKEY_OF, their_index.to_le_bytes().to_vec())
+    }
+
+    /// The session index this handshake declares it replaces, if any.
+    ///
+    /// A present-but-malformed marker is an **error**, not an absence. Reading a
+    /// truncated marker as "not a rekey" would reintroduce the silent
+    /// fall-through this field exists to remove.
+    pub fn rekey_of(&self) -> Result<Option<SessionIndex>, Error> {
+        let Some(entry) = self
+            .tlv_entries
+            .iter()
+            .find(|e| e.field_num == TLV_REKEY_OF)
+        else {
+            return Ok(None);
+        };
+        let bytes: [u8; 4] = entry
+            .value
+            .as_slice()
+            .try_into()
+            .map_err(|_| Error::Malformed("rekey marker is not 4 bytes"))?;
+        Ok(Some(SessionIndex::from_le_bytes(bytes)))
+    }
+
     /// Encode to wire format.
     pub fn encode(&self) -> Vec<u8> {
         let mut buf = Vec::with_capacity(NEGOTIATION_HEADER_SIZE);