Make rekey.enabled mean "initiate rekeys" and nothing else

The responder half of the establish decision was gated on this node's own
node.rekey.enabled. With the rekey now declared in the msg3 negotiation
payload, that flag was the only thing that could divert a msg3 whose marker
matched the session we hold, and it diverted it to a msg2 resend while the
initiator had already installed its pending session and would cut over on its
own timer regardless. A pair configured with the flag true on one end and
false on the other therefore parted company at the initiator's cutover and
carried no traffic in either direction until the link-dead timer.

Drop the flag from that gate. The arm now turns on the sender's declaration
matching the session we hold, which is the only signal that was ever
authoritative for the decision, and the snapshot field it read is gone since
every initiator-side use reads the config directly.

Removing the gate exposed a second reading of the same flag. check_rekey
returned early when rekey was disabled, and the drain-expiry arm of the rekey
poll is the only thing that releases a demoted session and its index. A node
that does not initiate had never accepted a rekey before, so it never
accumulated a drain; now that it does, the early return would pin the previous
session and its allocator index forever. The flag rides into the core as
RekeyCfg::initiate instead: the polled cutover and the trigger are gated on it,
the drain is not, and the shell runs the cadence unconditionally.

The cutover arm has to stay gated. The peer snapshot carries no rekey role, so
a pending session stored as responder satisfies that arm exactly as one we
initiated does; ungating it would move a non-initiating node's cutover off the
peer's k-bit flip and onto the poll.

Covered by a two-ended test asserting the pair converges on the same session
indices under an asymmetric setting, a node-level test that a non-initiating
responder actually releases the demoted session and index once the drain window
expires, and a core test pinning which arms a non-initiating snapshot yields.
The existing msg2-resend test reached its arm only through the removed gate and
now reaches it through the health conjunct instead.

src/node/dataplane/peer_actions.rs

@@ -543,11 +543,14 @@ impl Node {
                     let _ = did_cutover;
                 }
                 PeerAction::CompleteDrain { peer: node_addr } => {
-                    // Initiator drain completion: the live authoritative
-                    // rekey-cadence path, routed here from `check_rekey` via
+                    // Drain completion: the live authoritative rekey-cadence
+                    // path, routed here from `check_rekey` via
                     // `route_rekey_cadence` → `PeerEvent::RekeyConsume`; the
                     // inline body survives only as `drain_peer_inline`, a
-                    // debug-assert release fallback. Extract the real previous
+                    // debug-assert release fallback. Reached in either rekey
+                    // role — a responder's cutover demotes a session the same
+                    // way an initiator's does — including on a node whose own
+                    // trigger is off. Extract the real previous
                     // index + transport_id under the peer borrow, drop the
                     // borrow, then run the cache_key cleanup (which takes
                     // &mut self for unregister_decrypt_worker_session).

src/node/handlers/handshake.rs

@@ -1409,7 +1409,6 @@ impl Node {
         // 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;
 
         // Resolve the sender's declaration against the session we hold. Matched
         // ONLY against the peer this handshake authenticated as, never through a
@@ -1437,7 +1436,6 @@ impl Node {
                 rekey_in_progress: existing_peer.rekey_in_progress(),
                 existing_msg2: existing_peer.handshake_msg2().map(|m| m.to_vec()),
                 different_link: existing_peer.link_id() != link_id,
-                rekey_enabled,
                 rekey_claim,
                 our_node_addr,
             },
@@ -1450,7 +1448,6 @@ impl Node {
                 rekey_in_progress: false,
                 existing_msg2: None,
                 different_link: false,
-                rekey_enabled,
                 rekey_claim,
                 our_node_addr,
             },

src/node/handlers/rekey.rs

@@ -37,12 +37,17 @@ impl Node {
     /// - If the initiator has a pending session, perform K-bit cutover
     /// - If the drain window has expired, clean up the previous session
     /// - If the rekey timer/counter fires, initiate a new handshake
+    ///
+    /// Runs whatever `node.rekey.enabled` says, because the drain in the middle
+    /// of that list is not initiation: a node that never initiates still accepts
+    /// its peers' rekeys, and each acceptance demotes a live session that only
+    /// this path releases. The flag rides into the core as `RekeyCfg::initiate`
+    /// and suppresses the two initiating arms there, so which arms a disabled
+    /// node runs stays one decision in one place rather than an early return
+    /// here and a rule there.
     pub(in crate::node) async fn check_rekey(&mut self) {
-        if !self.config().node.rekey.enabled {
-            return;
-        }
-
         let cfg = RekeyCfg {
+            initiate: self.config().node.rekey.enabled,
             after_secs: self.config().node.rekey.after_secs,
             after_messages: self.config().node.rekey.after_messages,
         };

src/node/tests/handshake.rs

@@ -1634,8 +1634,8 @@ async fn drive_rekey_to_msg3(initiator: &mut HsNode, responder: &mut HsNode) ->
 
 #[tokio::test]
 async fn test_msg3_dual_rekey_won_frees_index() {
-    // Both ends carry the rekey config: the initiator to fire its own trigger,
-    // the responder for the `rekey_enabled` gate in the classifier.
+    // Both ends carry the rekey config so each can fire its own trigger; the
+    // classifier itself reads no config, only the sender's declaration.
     let make_config = || {
         let mut c = Config::new();
         c.node.rekey.enabled = true;
@@ -1719,20 +1719,25 @@ async fn test_msg3_dual_rekey_won_frees_index() {
 
 #[tokio::test]
 async fn test_msg3_resend_msg2_frees_index() {
-    // A declared rekey into a responder that has rekey DISABLED. The marker
-    // matches, so this is unambiguously a rekey and not a crossing dial, and the
-    // responder's `rekey_enabled` gate sends it to the duplicate arm.
+    // A declared rekey landing on a peer whose link is no longer healthy. The
+    // marker matches, so this is unambiguously a rekey and not a crossing dial;
+    // the classifier's health conjunct is what sends it to the duplicate arm,
+    // and the shell must then free the msg1-allocated index and leave the
+    // existing session alone.
     //
     // A bare second handshake will not reach this arm: it declares no rekey, and
-    // an undeclared msg3 on a different link is a cross-connection, which
-    // `rekey_enabled` does not gate. That is what this test used to do, and every
-    // assertion below holds on the cross-connection arm too, so it passed while
-    // exercising something else entirely.
+    // an undeclared msg3 on a different link is a cross-connection, on which
+    // every assertion below also holds — so routing through it would exercise
+    // something else entirely while passing.
+    //
+    // The responder's own `rekey.enabled` is deliberately NOT the lever here.
+    // That flag governs whether this node initiates rekeys and has no say in
+    // whether it accepts one; using it to reach this arm is the divergence
+    // `test_asymmetric_rekey_config_converges` exists to forbid.
     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 = false;
+    let config = Config::new();
 
     let mut initiator = make_hs_node(init_config).await;
     let mut responder = make_hs_node(config).await;
@@ -1760,13 +1765,29 @@ async fn test_msg3_resend_msg2_frees_index() {
     let session_before = (before.our_index(), before.their_index());
 
     // The rekey's msg1 allocates a fresh index; its msg3 then frees it on the
-    // duplicate arm, because this responder does not do rekeys.
+    // duplicate arm.
     let msg3b = drive_rekey_to_msg3(&mut initiator, &mut responder).await;
     assert_eq!(
         responder.node.index_allocator.count(),
         baseline + 1,
         "the rekey msg1 allocated a fresh index"
     );
+
+    // The link goes quiet past the heartbeat threshold while the rekey is in
+    // flight, exactly as the tick loop would mark it. An unhealthy peer is not a
+    // rekey candidate, so the declared rekey falls through to the duplicate arm.
+    // Marked after the msg1/msg2 exchange, since nothing on the handshake path
+    // re-marks a peer connected but a later `touch` would.
+    responder
+        .node
+        .get_peer_mut(&peer_addr)
+        .unwrap()
+        .mark_stale();
+    assert!(
+        !responder.node.get_peer(&peer_addr).unwrap().is_healthy(),
+        "the peer must be unhealthy, or this reaches the rekey-responder arm"
+    );
+
     responder.node.handle_msg3(msg3b).await;
 
     assert_eq!(
@@ -1921,8 +1942,9 @@ async fn test_msg3_crypto_fail_disposes_leg_machine() {
 #[tokio::test]
 async fn test_msg3_rekey_respond_disposes_leg_machine() {
     // 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.
+    // the rekey marker. The initiator needs the rekey config to fire the
+    // trigger; the responder's copy only lets it run its own cadence, since the
+    // classifier reads no config.
     let mut init_config = Config::new();
     init_config.node.rekey.enabled = true;
     init_config.node.rekey.after_secs = 1;
@@ -2063,6 +2085,269 @@ async fn test_count_triggered_rekey_on_young_session_converges() {
     stop_hs(&mut responder).await;
 }
 
+/// An asymmetric `rekey.enabled` must not split the link: the two ends still
+/// come out of the peer's rekey holding the same session.
+///
+/// `rekey.enabled` says whether this node INITIATES rekeys. It says nothing
+/// about whether it accepts one, and it cannot: by the time a rekey msg3
+/// arrives the sender has already committed to the new session, and the
+/// declaration on the wire is the only authoritative signal about what the msg3
+/// is. A responder that read its own trigger config as permission to decline
+/// would keep the old session while the peer sends on the new one, and the link
+/// would carry nothing in either direction until the link-dead timer.
+///
+/// Every assertion here spans BOTH nodes, comparing each end's send index
+/// against the index the other end receives on. The divergent outcome leaves
+/// each node internally consistent, so a one-sided check cannot see it — which
+/// is exactly how the defect survived: `test_msg3_resend_msg2_frees_index`
+/// drove a declared rekey into a rekey-disabled responder and asserted only the
+/// responder's own state, so it passed on full divergence.
+#[tokio::test]
+async fn test_asymmetric_rekey_config_converges() {
+    let mut init_config = Config::new();
+    init_config.node.rekey.enabled = true;
+    init_config.node.rekey.after_secs = 1;
+    // The responder never initiates a rekey of its own. It must still accept the
+    // peer's, and must still cut over to it.
+    let mut resp_config = Config::new();
+    resp_config.node.rekey.enabled = false;
+
+    let mut initiator = make_hs_node(init_config).await;
+    let mut responder = make_hs_node(resp_config).await;
+
+    let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await;
+    responder.node.handle_msg3(msg3).await;
+    assert_eq!(responder.node.peer_count(), 1);
+
+    let initiator_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();
+
+    // Age the initiator's session well past the jitter band ([-15, +15] around
+    // `after_secs`) so its own trigger fires and the msg3 that arrives carries a
+    // matching marker.
+    initiator
+        .node
+        .get_peer_mut(&responder_addr)
+        .unwrap()
+        .test_backdate_session_established(std::time::Duration::from_secs(120));
+
+    let rekey_msg3 = drive_rekey_to_msg3(&mut initiator, &mut responder).await;
+    responder.node.handle_msg3(rekey_msg3).await;
+
+    assert!(
+        responder
+            .node
+            .get_peer(&initiator_addr)
+            .unwrap()
+            .pending_new_session()
+            .is_some(),
+        "the responder declined the peer's rekey on account of its own trigger \
+         config: no pending session"
+    );
+
+    // The initiator cuts over on its own cadence and starts sending on the new
+    // session. Cross-node, before either cutover reaches the responder: the
+    // session the initiator now sends on has to be the one the responder is
+    // holding ready, and vice versa.
+    initiator.node.check_rekey().await;
+    {
+        let on_initiator = initiator.node.get_peer(&responder_addr).unwrap();
+        let on_responder = responder.node.get_peer(&initiator_addr).unwrap();
+        assert_eq!(
+            on_initiator.their_index(),
+            on_responder.pending_our_index(),
+            "the initiator sends on an index the responder is not standing by to \
+             receive on"
+        );
+        assert_eq!(
+            on_responder.pending_their_index(),
+            on_initiator.our_index(),
+            "the responder is standing by to send on an index the initiator does \
+             not receive on"
+        );
+        assert!(
+            on_initiator.our_index().is_some() && on_initiator.their_index().is_some(),
+            "both indices must be set, or the equalities above pass on None == None"
+        );
+    }
+
+    // The responder runs no rekey cadence at all here, so it cuts over the way a
+    // responder always does: on the peer's first frame carrying the flipped
+    // K-bit, which the data plane turns into exactly this call.
+    responder
+        .node
+        .get_peer_mut(&initiator_addr)
+        .unwrap()
+        .handle_peer_kbit_flip();
+
+    let on_initiator = initiator.node.get_peer(&responder_addr).unwrap();
+    let on_responder = responder.node.get_peer(&initiator_addr).unwrap();
+    assert_eq!(
+        on_initiator.their_index(),
+        on_responder.our_index(),
+        "after the rekey the ends hold different sessions: traffic drops one way"
+    );
+    assert_eq!(
+        on_responder.their_index(),
+        on_initiator.our_index(),
+        "after the rekey the ends hold different sessions: traffic drops one way"
+    );
+    assert!(
+        on_responder.our_index().is_some() && on_responder.their_index().is_some(),
+        "both indices must be set, or the equalities above pass on None == None"
+    );
+
+    stop_hs(&mut initiator).await;
+    stop_hs(&mut responder).await;
+}
+
+/// A node that never initiates a rekey still finishes the drain of one it
+/// accepted: the demoted session and its index come back.
+///
+/// Accepting a peer's rekey demotes our live session into the drain slot and
+/// keeps its index registered, so the frames still in flight on it decrypt. The
+/// drain arm of the rekey cadence is the only thing that ever releases either.
+/// Skipping the whole cadence on a node with `rekey.enabled = false` therefore
+/// pins one session and one index per accepted rekey, for the life of the peer —
+/// which is a leak that grows with every rekey the peer initiates.
+///
+/// Asserted against the peer's real state and the real allocator, not a log
+/// line: the previous session, the drain flag, the index-to-peer map and the
+/// allocator's live count all have to come back to where they started.
+#[tokio::test]
+async fn test_non_initiating_responder_completes_rekey_drain() {
+    let mut init_config = Config::new();
+    init_config.node.rekey.enabled = true;
+    init_config.node.rekey.after_secs = 1;
+    let mut resp_config = Config::new();
+    resp_config.node.rekey.enabled = false;
+
+    let mut initiator = make_hs_node(init_config).await;
+    let mut responder = make_hs_node(resp_config).await;
+
+    let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await;
+    responder.node.handle_msg3(msg3).await;
+    assert_eq!(responder.node.peer_count(), 1);
+    let baseline = responder.node.index_allocator.count();
+    assert_eq!(baseline, 1, "responder holds exactly the peer's index");
+
+    let initiator_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));
+
+    // The peer's rekey completes and the responder accepts it, exactly as
+    // `test_asymmetric_rekey_config_converges` establishes it must.
+    let rekey_msg3 = drive_rekey_to_msg3(&mut initiator, &mut responder).await;
+    responder.node.handle_msg3(rekey_msg3).await;
+
+    let (old_index, transport) = {
+        let peer = responder.node.get_peer(&initiator_addr).unwrap();
+        assert!(
+            peer.pending_new_session().is_some(),
+            "the responder did not accept the rekey, so there is no drain to test"
+        );
+        (peer.our_index().unwrap(), peer.transport_id().unwrap())
+    };
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline + 1,
+        "the accepted rekey holds a second index"
+    );
+
+    // The responder cuts over on the peer's first frame on the new epoch, which
+    // demotes the session it was using into the drain slot.
+    responder
+        .node
+        .get_peer_mut(&initiator_addr)
+        .unwrap()
+        .handle_peer_kbit_flip();
+
+    {
+        let peer = responder.node.get_peer(&initiator_addr).unwrap();
+        assert!(peer.is_draining(), "the cutover must start a drain window");
+        assert!(
+            peer.previous_session().is_some(),
+            "the demoted session is kept for in-flight frames"
+        );
+        assert_eq!(peer.previous_our_index(), Some(old_index));
+        assert_ne!(
+            peer.our_index(),
+            Some(old_index),
+            "the live session must have moved off the drained index"
+        );
+    }
+    assert!(
+        responder
+            .node
+            .peers_by_index
+            .contains_key(&(transport, old_index.as_u32())),
+        "the drained index stays routable while the window is open"
+    );
+
+    // Mid-window the cadence must leave it alone, or the assertion below would
+    // pass on an arm that fires unconditionally.
+    responder.node.check_rekey().await;
+    assert!(
+        responder
+            .node
+            .get_peer(&initiator_addr)
+            .unwrap()
+            .previous_session()
+            .is_some(),
+        "the drain completed before its window expired"
+    );
+
+    // Window expires. This is the only path that releases the demoted session,
+    // and a node that does not initiate rekeys still has to run it.
+    responder
+        .node
+        .get_peer_mut(&initiator_addr)
+        .unwrap()
+        .test_backdate_drain_start(std::time::Duration::from_secs(600));
+    responder.node.check_rekey().await;
+
+    let peer = responder.node.get_peer(&initiator_addr).unwrap();
+    assert!(
+        peer.previous_session().is_none(),
+        "the demoted session is pinned forever on a node that never initiates"
+    );
+    assert!(!peer.is_draining(), "the drain window must be closed");
+    assert_eq!(peer.previous_our_index(), None);
+    assert!(
+        !peer.rekey_in_progress(),
+        "a node that does not initiate must not have started a rekey of its own"
+    );
+    assert!(
+        peer.has_session() && peer.our_index().is_some(),
+        "the live session must survive the drain"
+    );
+    assert!(
+        !responder
+            .node
+            .peers_by_index
+            .contains_key(&(transport, old_index.as_u32())),
+        "the drained index is still routable, so it was never released"
+    );
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline,
+        "the drained index never came back to the allocator"
+    );
+    assert_eq!(responder.node.peer_count(), 1, "active peer untouched");
+
+    stop_hs(&mut initiator).await;
+    stop_hs(&mut responder).await;
+}
+
 /// A fresh dial crossing our own in-flight rekey leaves BOTH ends on the same
 /// session, and leaks nothing on the way.
 ///

src/peer/active.rs

@@ -1040,6 +1040,18 @@ impl ActivePeer {
             .unwrap_or_else(Instant::now);
     }
 
+    /// Test-only seam: backdate the drain-start instant so a test can make
+    /// `drain_expired()` read true without waiting out the drain window. Only
+    /// shifts the private timestamp field, and is a no-op when no drain is in
+    /// progress, so it cannot manufacture a drain that did not happen. It changes
+    /// no decision logic, no threshold, and is compiled out of release builds.
+    #[cfg(test)]
+    pub(crate) fn test_backdate_drain_start(&mut self, age: std::time::Duration) {
+        if let Some(started) = self.send.drain_started {
+            self.send.drain_started = Some(started.checked_sub(age).unwrap_or_else(Instant::now));
+        }
+    }
+
     /// Per-session symmetric rekey-timer jitter offset (seconds).
     ///
     /// Drawn at session construction and at each rekey cutover; uniform

src/peer/machine.rs

@@ -1829,6 +1829,9 @@ impl PeerMachine {
             None => return Vec::new(),
         };
         let cfg = RekeyCfg {
+            // The machine's shadow cadence sees no node config and has always run
+            // the full arm set, so it keeps initiating.
+            initiate: true,
             after_secs: REKEY_AFTER_SECS,
             after_messages: REKEY_AFTER_MESSAGES,
         };
@@ -2271,7 +2274,6 @@ mod tests {
             rekey_in_progress: false,
             existing_msg2: None,
             different_link: false,
-            rekey_enabled: true,
             rekey_claim: RekeyClaim::None,
             our_node_addr: our,
         }
@@ -3201,11 +3203,13 @@ mod tests {
         let seed = SessionIndex::new(0xEF);
         let stored_msg2 = vec![1u8, 2, 3, 4];
 
-        // Same epoch, same link, rekey disabled -> duplicate handshake.
+        // Same epoch, same link, no rekey declared -> duplicate handshake. The
+        // `est_new_peer` default of `RekeyClaim::None` plus `different_link =
+        // false` is what pins this arm: an undeclared msg3 on the peer's own
+        // link is neither a rekey nor a crossing dial.
         let mut est = est_new_peer(our);
         est.has_existing_peer = true;
         est.existing_peer_epoch = Some([9u8; 8]);
-        est.rekey_enabled = false;
         est.existing_msg2 = Some(stored_msg2.clone());
         let wire = wire_outcome(peer_addr, Some([9u8; 8]));
 

src/proto/fmp/core.rs

@@ -186,8 +186,16 @@ pub(crate) struct RekeyResendSnapshot {
     pub msg1: Vec<u8>,
 }
 
-/// The rekey trigger thresholds, read shell-side from node config.
+/// The rekey cadence config, read shell-side from node config.
 pub(crate) struct RekeyCfg {
+    /// This node initiates rekeys (`node.rekey.enabled`).
+    ///
+    /// It governs the initiating half of the cadence only: the trigger, and the
+    /// polled cutover that follows an initiation. Finishing the drain of a rekey
+    /// already accepted as responder is housekeeping rather than initiation, and
+    /// runs whatever this says — a node that never initiates still has drains to
+    /// complete, because a peer's rekey it accepted demoted a live session.
+    pub initiate: bool,
     /// Rekey after this many seconds of session age (before jitter).
     pub after_secs: u64,
     /// Rekey after this many sent messages.
@@ -251,9 +259,6 @@ 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 a declared
-    /// rekey as a rekey rather than a duplicate).
-    pub rekey_enabled: bool,
     /// 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,
@@ -631,22 +636,36 @@ impl Fmp {
     /// Actions are returned phase-grouped (all cutovers, then all drains, then
     /// all rekey initiations) to preserve the pre-refactor global execution
     /// order across peers, which the shared `index_allocator` observes.
+    ///
+    /// `cfg.initiate` splits the cadence in two. The trigger and the polled
+    /// cutover belong to initiation and are suppressed when it is off; the drain
+    /// arm is housekeeping for a demoted session and is not. A node that only
+    /// ever responds to rekeys still accumulates drains, and nothing else
+    /// completes one.
     pub(crate) fn poll_rekey(&self, peers: Vec<PeerSnapshot>, cfg: &RekeyCfg) -> Vec<ConnAction> {
         let mut cutovers = Vec::new();
         let mut drains = Vec::new();
         let mut rekeys = Vec::new();
         for p in peers {
-            // 1. Initiator-side cutover.
-            if p.has_pending && !p.rekey_in_progress {
+            // 1. Polled cutover.
+            //
+            // Gated on `initiate`, and the gate is doing real work: the snapshot
+            // carries no rekey role, so this arm fires for a pending session we
+            // stored as RESPONDER just as readily as for one we initiated. On a
+            // node that does not initiate, the responder's cutover is the peer's
+            // K-bit flip on the data path, and polling it here would move that
+            // cutover earlier — a behaviour change this arm must not make.
+            if cfg.initiate && p.has_pending && !p.rekey_in_progress {
                 cutovers.push(ConnAction::Cutover { peer: p.addr });
                 continue;
             }
             // 2. Drain window expiry (does not preclude a trigger below).
+            // Deliberately ungated: see `RekeyCfg::initiate`.
             if p.is_draining && p.drain_expired {
                 drains.push(ConnAction::Drain { peer: p.addr });
             }
             // 3. Rekey trigger.
-            if p.rekey_in_progress || p.is_dampened || p.rekey_msg3_pending {
+            if !cfg.initiate || p.rekey_in_progress || p.is_dampened || p.rekey_msg3_pending {
                 continue;
             }
             let effective_after = cfg.after_secs.saturating_add_signed(p.jitter_secs);
@@ -725,8 +744,11 @@ impl Fmp {
     ///    this tie-break cannot see that state, so declining here would diverge
     ///    the pair; the executor abandons the displaced rekey instead.
     /// 5. Same epoch, marker naming the session we hold
-    ///    ([`Matches`](RekeyClaim::Matches)), rekey enabled, healthy session →
-    ///    [`RekeyRespond`]. The widened dual-init tie-break fires when the peer
+    ///    ([`Matches`](RekeyClaim::Matches)), healthy session → [`RekeyRespond`],
+    ///    whatever the local `rekey.enabled` says: that flag governs whether we
+    ///    initiate rekeys, never whether we accept one, or an asymmetric setting
+    ///    would diverge the two ends of the link.
+    ///    The widened dual-init tie-break fires when the peer
     ///    is in *either* the `rekey_in_progress` or the `pending_new_session`
     ///    state: the smaller NodeAddr wins ([`Reject`] the peer's `msg3`), the
     ///    larger loses (`abandon_first`, then respond).
@@ -810,11 +832,16 @@ impl Fmp {
                 }
 
                 // 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
-                {
+                //
+                // Nothing here consults the local rekey config. `rekey.enabled`
+                // means "initiate rekeys" and nothing else: reading it here made
+                // the responder decline a rekey the sender had already committed
+                // to, so an asymmetric setting left the two ends on different
+                // sessions with the link carrying nothing until the link-dead
+                // timer. The sender's declaration matching the session we hold is
+                // the authoritative signal, and it is a property of the wire, not
+                // of our config.
+                if snap.rekey_claim == RekeyClaim::Matches && snap.has_session && snap.is_healthy {
                     // Widened dual-init tie-break: both the still-in-progress and
                     // the already-pending states resolve by the smaller NodeAddr.
                     if snap.rekey_in_progress || snap.pending_new_session {

src/proto/fmp/tests/core.rs

@@ -15,15 +15,25 @@ use crate::transport::LinkId;
 /// Matching-epoch default used by `establish_snapshot`.
 const SAME_EPOCH: [u8; 8] = [0x01; 8];
 
-/// Threshold config used across the rekey decision tests: rekey at 100s of
-/// session age or 1000 sent messages.
+/// Threshold config used across the rekey decision tests: an initiating node
+/// that rekeys at 100s of session age or 1000 sent messages.
 fn cfg() -> RekeyCfg {
     RekeyCfg {
+        initiate: true,
         after_secs: 100,
         after_messages: 1_000,
     }
 }
 
+/// The same thresholds on a node that does not initiate rekeys
+/// (`node.rekey.enabled = false`).
+fn cfg_no_initiate() -> RekeyCfg {
+    RekeyCfg {
+        initiate: false,
+        ..cfg()
+    }
+}
+
 #[test]
 fn empty_stale_set_yields_no_actions() {
     let fmp = Fmp::new();
@@ -161,6 +171,50 @@ fn rekey_expired_drain_and_trigger_both_fire() {
     );
 }
 
+/// A node that does not initiate rekeys still completes an expired drain, and
+/// still does nothing else.
+///
+/// The two halves are one property. A node with `rekey.enabled = false` accepts
+/// its peers' rekeys — the declaration on the wire decides that, not local
+/// config — and every acceptance demotes a live session into the drain slot.
+/// Nothing but this arm releases that session and its index, so suppressing the
+/// whole poll strands both for the life of the peer.
+///
+/// The other half is why the suppression cannot simply be dropped: this snapshot
+/// carries no rekey role, so `has_pending` is set by a responder-stored session
+/// exactly as by an initiated one. Letting the cutover arm run here would move a
+/// non-initiating node's cutover off the peer's K-bit flip and onto the poll.
+#[test]
+fn rekey_without_initiation_drains_but_does_not_cut_over_or_trigger() {
+    let fmp = Fmp::new();
+
+    let mut draining = peer_snapshot(0x14);
+    draining.is_draining = true;
+    draining.drain_expired = true;
+    // Also over both trigger thresholds, and holding a pending session as a
+    // responder would: neither may produce an action.
+    draining.has_pending = true;
+    draining.elapsed_secs = 10_000;
+    draining.counter = 10_000;
+    let actions = fmp.poll_rekey(vec![draining], &cfg_no_initiate());
+    assert_eq!(
+        actions.len(),
+        1,
+        "a non-initiating node must emit the drain and nothing else"
+    );
+    assert!(matches!(actions[0], ConnAction::Drain { peer } if peer == make_node_addr(0x14)));
+
+    // An unexpired drain window is still not due, so the arm is not simply
+    // firing on every poll.
+    let mut unexpired = peer_snapshot(0x14);
+    unexpired.is_draining = true;
+    assert!(
+        fmp.poll_rekey(vec![unexpired], &cfg_no_initiate())
+            .is_empty(),
+        "an unexpired drain window must not complete"
+    );
+}
+
 /// The send-counter arm of the rekey trigger, pinned at its boundary through
 /// the real `poll_rekey` rather than a reimplementation of its predicate.
 ///
@@ -423,23 +477,15 @@ fn establish_same_link_fresh_is_duplicate() {
     }
 }
 
-/// Aged same-epoch session, rekey disabled: not a rekey -> duplicate resend
-/// (mirrors the rekey-disabled chartest).
-#[test]
-fn establish_aged_rekey_disabled_is_duplicate() {
-    let fmp = Fmp::new();
-    let mut snap = establish_snapshot(0x05);
-    snap.rekey_enabled = false;
-    snap.rekey_claim = RekeyClaim::Matches; // declared, but rekey disabled
-    let wire = wire_outcome(0x02, SAME_EPOCH);
-    assert!(matches!(
-        fmp.establish_inbound(&snap, &wire),
-        InboundDecision::ResendMsg2 { .. }
-    ));
-}
-
 /// Aged, healthy, same-epoch session with no dual-init in flight -> rekey
 /// responder, no prior abandon.
+///
+/// The snapshot carries no local rekey-trigger config, and deliberately: the
+/// sender's declaration is the whole input. A responder that consulted its own
+/// `rekey.enabled` here would decline a rekey the sender has already committed
+/// to, leaving the two ends on different sessions whenever the setting is
+/// asymmetric. The two-ended proof of that lives in `node/tests/handshake.rs`;
+/// this pins the decision itself.
 #[test]
 fn establish_aged_rekey_responds_without_abandon() {
     let fmp = Fmp::new();

src/proto/fmp/tests/util.rs

@@ -70,8 +70,8 @@ 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, owned by node `our_byte`.
+/// Build an `EstablishSnapshot` describing an existing, healthy, same-epoch
+/// peer owned by node `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
@@ -86,7 +86,6 @@ pub(super) fn establish_snapshot(our_byte: u8) -> EstablishSnapshot {
         rekey_in_progress: false,
         existing_msg2: None,
         different_link: false,
-        rekey_enabled: true,
         rekey_claim: RekeyClaim::None,
         our_node_addr: make_node_addr(our_byte),
     }