diff --git a/src/control/queries.rs b/src/control/queries.rs
index 4e1089b..7908ab5 100644
--- a/src/control/queries.rs
+++ b/src/control/queries.rs
@@ -1323,7 +1323,7 @@ pub fn show_connections(node: &Node) -> Value {
                 "handshake_state": format!("{}", conn.handshake_state()),
                 "started_at_ms": conn.started_at(),
                 "idle_ms": now.saturating_sub(conn.last_activity()),
-                "resend_count": conn.resend_count(),
+                "resend_count": node.connection_resend_count(conn.link_id()),
             });
 
             if let Some(identity) = conn.expected_identity() {
diff --git a/src/node/dataplane/dispatch.rs b/src/node/dataplane/dispatch.rs
index 4949d5b..f63676d 100644
--- a/src/node/dataplane/dispatch.rs
+++ b/src/node/dataplane/dispatch.rs
@@ -191,7 +191,7 @@ impl Node {
         // derived above (a peer's link_id is immutable, so the key never
         // moved). Keeps peers <-> peer_machines in exact correspondence on
         // teardown. NEUTRAL: nothing reads peer_machines yet.
-        self.peer_machines.remove(&link_id);
+        self.remove_peer_machine(link_id);
         if let Some(transport_id) = transport_id {
             self.cleanup_bootstrap_transport_if_unused(transport_id);
         }
diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs
index fab03e3..5ea6b5c 100644
--- a/src/node/dataplane/peer_actions.rs
+++ b/src/node/dataplane/peer_actions.rs
@@ -21,9 +21,14 @@
 //! `SendHandshake` `their_index == Some` (msg2) branch stays dormant.
 //!
 //! Not yet driven, so their arms stay inert stubs: rekey/crypto installs,
-//! link-control frames, the timers (`SetTimer`/`CancelTimer` — the legacy tick
-//! still runs them), and the connected-UDP plane. `RegisterDecryptSession` is a
-//! deliberate no-op — see its arm for the note.
+//! link-control frames, and the connected-UDP plane. `RegisterDecryptSession` is
+//! a deliberate no-op — see its arm for the note.
+//!
+//! The timer arms (`SetTimer`/`CancelTimer`) populate/clear the per-peer timer
+//! store (`peer_timers`). The `HandshakeRetransmit` and `HandshakeTimeout`
+//! deadlines are read and fired by `drive_peer_timers` (the handshake resend +
+//! reap home). The rekey/liveness kinds are still SHADOW — driven by their own
+//! shell drivers — so populating them stays behavior-neutral.
 
 use crate::PeerIdentity;
 use crate::node::reject::{HandshakeReject, RejectReason};
@@ -149,7 +154,7 @@ impl Node {
                             Err(_e) => {
                                 self.links.remove(&link);
                                 self.addr_to_link.remove(&(transport_id, remote_addr));
-                                self.peer_machines.remove(&link);
+                                self.remove_peer_machine(link);
                                 return;
                             }
                         }
@@ -194,7 +199,7 @@ impl Node {
                             if let Some(idx) = ambient.our_index {
                                 let _ = self.index_allocator.free(idx);
                             }
-                            self.peer_machines.remove(&link);
+                            self.remove_peer_machine(link);
                             self.stats_mut()
                                 .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
                             return;
@@ -572,10 +577,21 @@ impl Node {
                     // Connected-UDP plane ownership (`connected_udp.rs`). Out of
                     // scope for now.
                 }
-                PeerAction::SetTimer { .. } | PeerAction::CancelTimer { .. } => {
-                    // Timers become actions on the existing quantized tick. INERT:
-                    // the legacy tick timers still run, so driving these would
-                    // double-schedule.
+                PeerAction::SetTimer { kind, at_ms } => {
+                    // Populate the per-peer timer store (overwrite = reschedule).
+                    // The `HandshakeRetransmit` and `HandshakeTimeout` deadlines
+                    // are read + fired by `drive_peer_timers`. Rekey/liveness kinds
+                    // are still SHADOW here — they keep their own shell drivers —
+                    // so populating them stays behavior-neutral.
+                    self.peer_timers
+                        .entry(link)
+                        .or_default()
+                        .insert(kind, at_ms);
+                }
+                PeerAction::CancelTimer { kind } => {
+                    if let Some(timers) = self.peer_timers.get_mut(&link) {
+                        timers.remove(&kind);
+                    }
                 }
                 PeerAction::ReportLost { peer, kind } => {
                     // The single loss token, routed to the reconciler reflex the
diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs
index 4ebf5a5..19fd7f0 100644
--- a/src/node/dataplane/rx_loop.rs
+++ b/src/node/dataplane/rx_loop.rs
@@ -336,7 +336,7 @@ impl Node {
                     self.poll_pending_connects().await;
                     self.poll_nostr_rendezvous().await;
                     self.poll_lan_rendezvous().await;
-                    self.resend_pending_handshakes(now_ms).await;
+                    self.drive_peer_timers(now_ms).await;
                     self.resend_pending_rekeys(now_ms).await;
                     self.resend_pending_fmp_rekey_msg3(now_ms).await;
                     self.resend_pending_session_handshakes(now_ms).await;
diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs
index 78fea02..8492d72 100644
--- a/src/node/handlers/handshake.rs
+++ b/src/node/handlers/handshake.rs
@@ -10,7 +10,7 @@ use crate::node::acl::PeerAclContext;
 use crate::node::dataplane::PeerActionCtx;
 use crate::node::reject::{HandshakeReject, RejectReason};
 use crate::node::{Node, NodeError};
-use crate::peer::machine::{PeerAction, PeerEvent, PeerMachine};
+use crate::peer::machine::{PeerAction, PeerEvent, PeerMachine, TimerKind};
 use crate::peer::{ActivePeer, PeerConnection};
 use crate::proto::fmp::wire::{Msg1Header, Msg2Header, Msg3Header, build_msg2, build_msg3};
 use crate::proto::fmp::{
@@ -598,7 +598,7 @@ impl Node {
             self.pending_outbound.remove(&key);
             self.connections.remove(&link_id);
             // Drop the machine persisted at dial — this leg never promotes.
-            self.peer_machines.remove(&link_id);
+            self.remove_peer_machine(link_id);
             self.remove_link(&link_id);
             self.stats_mut()
                 .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
@@ -667,7 +667,7 @@ impl Node {
             // directly, with no machine). Drop it on entry so none of this block's
             // exits leave a dangling machine — matching the pre-persistence path,
             // which created no machine for a cross-connection.
-            self.peer_machines.remove(&link_id);
+            self.remove_peer_machine(link_id);
             let our_outbound_wins = cross_connection_winner(
                 self.identity().node_addr(),
                 &peer_node_addr,
@@ -827,9 +827,22 @@ impl Node {
                 )
             }
         };
+        // The outbound Msg2 Promote step cancels the two dial-armed handshake
+        // timers (the machine survives promotion, so they would otherwise linger
+        // in the driver's shadow store) and then promotes. The cancels are
+        // behavior-neutral at this rung — `peer_timers` is written but not yet
+        // driven — and `PromoteToActive` is still what performs the promotion.
         debug_assert_eq!(
             promote_actions,
-            vec![PeerAction::PromoteToActive { link: link_id }]
+            vec![
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeRetransmit
+                },
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeTimeout
+                },
+                PeerAction::PromoteToActive { link: link_id },
+            ]
         );
 
         let ambient = PeerActionCtx {
@@ -1443,7 +1456,7 @@ impl Node {
                 // drop it so no machine orphans when its ActivePeer is removed.
                 // The winning connection's machine is inserted below keyed by
                 // the winner link_id. NEUTRAL: nothing reads peer_machines yet.
-                self.peer_machines.remove(&loser_link_id);
+                self.remove_peer_machine(loser_link_id);
 
                 // Clean up old peer's index from peers_by_index
                 if let (Some(old_tid), Some(old_idx)) =
diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs
index 2aa3ff7..f1a6b79 100644
--- a/src/node/handlers/timeout.rs
+++ b/src/node/handlers/timeout.rs
@@ -2,7 +2,7 @@
 //! and handshake message resend scheduling.
 
 use crate::node::Node;
-use crate::peer::HandshakeState;
+use crate::peer::machine::TimerKind;
 use crate::proto::fmp::{
     ConnAction, ConnSnapshot, LifecycleView, PeerSnapshot, RekeyResendSnapshot,
 };
@@ -11,9 +11,21 @@ use tracing::{debug, info};
 
 impl LifecycleView for Node {
     fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec<ConnSnapshot> {
+        // `is_failed()` legs are always reaped here (~1s), as before. The
+        // idle-timeout is reaped here ONLY for legs whose timeout is not already
+        // driven by a machine `HandshakeTimeout` timer — i.e. inbound legs (IK
+        // inbound arms none) and machine-less test connections. Outbound legs with
+        // an armed timer are reaped by `drive_handshake_timeouts`, so excluding
+        // them here avoids a double reap.
         self.connections
             .iter()
-            .filter(|(_, conn)| conn.is_timed_out(now_ms, timeout_ms) || conn.is_failed())
+            .filter(|(link_id, conn)| {
+                conn.is_failed()
+                    || (conn.is_timed_out(now_ms, timeout_ms)
+                        && !self.peer_timers.get(*link_id).is_some_and(|timers| {
+                            timers.contains_key(&TimerKind::HandshakeTimeout)
+                        }))
+            })
             .map(|(link_id, conn)| ConnSnapshot {
                 link: *link_id,
                 is_outbound: conn.is_outbound(),
@@ -24,36 +36,6 @@ impl LifecycleView for Node {
             .collect()
     }
 
-    fn resend_candidates(&self, now_ms: u64, max_resends: u32) -> Vec<ConnSnapshot> {
-        self.connections
-            .iter()
-            .filter(|(_, conn)| {
-                conn.is_outbound()
-                    && conn.handshake_state() == HandshakeState::SentMsg1
-                    && conn.resend_count() < max_resends
-                    && conn.next_resend_at_ms() > 0
-                    && now_ms >= conn.next_resend_at_ms()
-                    // Skip resend if the target peer is already promoted — a
-                    // cross-connection was resolved via the inbound path and
-                    // resending msg1 would start a new handshake on the peer,
-                    // creating a session mismatch.
-                    && !conn
-                        .expected_identity()
-                        .map(|id| self.peers.contains_key(id.node_addr()))
-                        .unwrap_or(false)
-            })
-            .filter_map(|(link_id, conn)| {
-                conn.handshake_msg1().map(|msg1| ConnSnapshot {
-                    link: *link_id,
-                    is_outbound: true,
-                    retry_addr: None,
-                    resend_count: conn.resend_count(),
-                    msg1: msg1.to_vec(),
-                })
-            })
-            .collect()
-    }
-
     fn rekey_peers(&self) -> Vec<PeerSnapshot> {
         // The snapshot builder lives in `rekey` beside its drain/dampening
         // constants; the read-seam unifies here.
@@ -127,7 +109,7 @@ impl Node {
         // `link_id` and lifetime; drop it here so a reaped handshake leg leaves
         // no dangling machine. A no-op for promoted peers — `promote_connection`
         // already removed their connection, so this reaper never runs for them.
-        self.peer_machines.remove(&link_id);
+        self.remove_peer_machine(link_id);
         let transport_id = conn.transport_id();
 
         // Free session index and pending_outbound/pending_inbound if allocated
@@ -146,22 +128,161 @@ impl Node {
         }
     }
 
-    /// Resend handshake messages for pending connections.
+    /// Act on the per-peer machine timers this tick.
     ///
-    /// For outbound connections in SentMsg1 state, resends the stored msg1
-    /// with exponential backoff. Called periodically from the RX event loop.
-    pub(in crate::node) async fn resend_pending_handshakes(&mut self, now_ms: u64) {
-        if self.connections.is_empty() {
+    /// The sans-IO machine arms `SetTimer`/`CancelTimer` actions into
+    /// [`peer_timers`](Node::peer_timers); this is the shell driver that acts on
+    /// them: timeout reaps idle-timed-out outbound legs, retransmit resends the
+    /// due msg1s. Handshake-TIMEOUT is driven before handshake-RETRANSMIT so a
+    /// timed-out leg is reaped rather than resent on the same tick. The
+    /// rekey/liveness kinds keep their own shell drivers, so only the two
+    /// handshake kinds are driven here.
+    pub(in crate::node) async fn drive_peer_timers(&mut self, now_ms: u64) {
+        if self.peer_timers.is_empty() {
             return;
         }
+        self.drive_handshake_timeouts(now_ms);
+        self.drive_handshake_retransmits(now_ms).await;
+    }
 
+    /// Reap the outbound legs whose machine `HandshakeTimeout` timer marks them
+    /// as machine-timeout-owned and which have idle-timed-out this tick.
+    ///
+    /// The timer's PRESENCE selects the leg (only OUTBOUND legs arm one — IK
+    /// inbound arms none); the reap THRESHOLD is the shell `is_timed_out(now,
+    /// config)` predicate, NOT the timer's stored deadline. This matters because
+    /// the machine arms the timer from a hardcoded constant at dial, which is not
+    /// authoritative for an operator-tuned `handshake_timeout_secs` — reading the
+    /// threshold from config each tick keeps the reap neutral for any config, and
+    /// off the `last_activity` clock exactly as the old `check_timeouts` did. A
+    /// timed-out leg is reaped by the old Teardown path: the outbound retry
+    /// reflex, then `cleanup_stale_connection` (which drops the machine + timers).
+    ///
+    /// `check_timeouts` keeps reaping everything else — `is_failed()` legs and the
+    /// idle-timeout of legs without a machine timer (inbound / machine-less).
+    fn drive_handshake_timeouts(&mut self, now_ms: u64) {
+        let timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000;
+        let timer_links: Vec<LinkId> = self
+            .peer_timers
+            .iter()
+            .filter(|(_, timers)| timers.contains_key(&TimerKind::HandshakeTimeout))
+            .map(|(link, _)| *link)
+            .collect();
+        for link in timer_links {
+            let (reap, retry_peer) = match self.connections.get(&link) {
+                Some(conn) if conn.is_timed_out(now_ms, timeout_ms) => {
+                    let retry_peer = if conn.is_outbound() {
+                        conn.expected_identity().map(|id| *id.node_addr())
+                    } else {
+                        None
+                    };
+                    (true, retry_peer)
+                }
+                // Not yet idle-timed-out: leave the timer for a later tick.
+                Some(_) => (false, None),
+                None => {
+                    // Orphan timer (connection already reaped elsewhere) — drop it.
+                    if let Some(timers) = self.peer_timers.get_mut(&link) {
+                        timers.remove(&TimerKind::HandshakeTimeout);
+                    }
+                    (false, None)
+                }
+            };
+            if reap {
+                if let Some(peer) = retry_peer {
+                    self.note_handshake_timeout(peer, now_ms);
+                }
+                debug!(link_id = %link, "Handshake connection timed out");
+                self.cleanup_stale_connection(link, now_ms);
+            }
+        }
+    }
+
+    /// Fire due handshake-retransmit timers: resend the stored msg1.
+    ///
+    /// The pre-fold `resend_pending_handshakes` logic, re-homed: the *due* signal
+    /// is the machine-armed timer (not the connection's `next_resend_at_ms`), and
+    /// the resend counter lives on the machine (the operator-visible count reads
+    /// from there). The wire bytes and transport target still come from the shell
+    /// connection, the pure core computes the backoff schedule, and — matching the
+    /// old shell exactly — the count and reschedule advance only on a successful
+    /// send; a failed send neither advances the count nor marks the connection
+    /// failed, it just retries next tick.
+    async fn drive_handshake_retransmits(&mut self, now_ms: u64) {
         let max_resends = self.config().node.rate_limit.handshake_max_resends;
         let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms;
         let backoff = self.config().node.rate_limit.handshake_resend_backoff;
 
-        // The shell resolves the resend-candidate predicate and copies the
-        // opaque msg1 bytes; the core computes the backoff schedule.
-        let candidates = self.resend_candidates(now_ms, max_resends);
+        // Collect due retransmit timers (kind-filtered).
+        let due: Vec<LinkId> = self
+            .peer_timers
+            .iter()
+            .filter(|(_, timers)| {
+                timers
+                    .get(&TimerKind::HandshakeRetransmit)
+                    .is_some_and(|&at_ms| now_ms >= at_ms)
+            })
+            .map(|(link, _)| *link)
+            .collect();
+        if due.is_empty() {
+            return;
+        }
+
+        // Classify each due link against the machine + connection. A timer whose
+        // machine has left `SentMsg1` (promoted/gone) or has hit the resend cap
+        // is dropped — no more resends, exactly as the old shell stopped
+        // selecting a capped/settled connection; the handshake-timeout reaper
+        // takes it from there.
+        let mut candidates: Vec<ConnSnapshot> = Vec::new();
+        let mut drop_timers: Vec<LinkId> = Vec::new();
+        for link in due {
+            // Skip a resend whose target peer already promoted via the inbound
+            // cross-connection path — resending msg1 would start a new handshake
+            // (session mismatch).
+            if let Some(conn) = self.connections.get(&link)
+                && conn
+                    .expected_identity()
+                    .map(|id| self.peers.contains_key(id.node_addr()))
+                    .unwrap_or(false)
+            {
+                drop_timers.push(link);
+                continue;
+            }
+            let armed = match self.peer_machines.get(&link) {
+                Some(machine)
+                    if machine.is_handshaking_sent_msg1()
+                        && machine.resend_count() < max_resends =>
+                {
+                    machine.resend_count()
+                }
+                Some(_) => {
+                    drop_timers.push(link);
+                    continue;
+                }
+                None => {
+                    drop_timers.push(link);
+                    continue;
+                }
+            };
+            match self.connections.get(&link).and_then(|c| c.handshake_msg1()) {
+                // Armed but the stored wire isn't there yet — leave the timer and
+                // retry next tick (matches the old candidate filter skipping it).
+                None => continue,
+                Some(msg1) => candidates.push(ConnSnapshot {
+                    link,
+                    is_outbound: true,
+                    retry_addr: None,
+                    resend_count: armed,
+                    msg1: msg1.to_vec(),
+                }),
+            }
+        }
+        for link in drop_timers {
+            if let Some(timers) = self.peer_timers.get_mut(&link) {
+                timers.remove(&TimerKind::HandshakeRetransmit);
+            }
+        }
+
         for action in self
             .fmp
             .poll_resends(candidates, now_ms, interval_ms, backoff)
@@ -175,7 +296,6 @@ impl Node {
                 continue;
             };
 
-            // Get transport and address info from the connection
             let (transport_id, remote_addr) = match self.connections.get(&link) {
                 Some(conn) => match (conn.transport_id(), conn.source_addr()) {
                     (Some(tid), Some(addr)) => (tid, addr.clone()),
@@ -184,7 +304,6 @@ impl Node {
                 None => continue,
             };
 
-            // Send the stored msg1
             let sent = if let Some(transport) = self.transports.get(&transport_id) {
                 match transport.send(&remote_addr, &bytes).await {
                     Ok(_) => true,
@@ -201,13 +320,26 @@ impl Node {
                 false
             };
 
-            if sent && let Some(conn) = self.connections.get_mut(&link) {
-                conn.record_resend(next_resend_at_ms);
-                debug!(
-                    link_id = %link,
-                    resend = conn.resend_count(),
-                    "Resent handshake msg1"
-                );
+            if sent {
+                if let Some(machine) = self.peer_machines.get_mut(&link) {
+                    machine.record_resend(next_resend_at_ms);
+                    debug!(
+                        link_id = %link,
+                        resend = machine.resend_count(),
+                        "Resent handshake msg1"
+                    );
+                }
+                self.peer_timers
+                    .entry(link)
+                    .or_default()
+                    .insert(TimerKind::HandshakeRetransmit, next_resend_at_ms);
+            } else {
+                // Failed send: keep retrying at the tick cadence (the old shell
+                // left next_resend_at_ms unchanged so the connection stayed due).
+                self.peer_timers
+                    .entry(link)
+                    .or_default()
+                    .insert(TimerKind::HandshakeRetransmit, now_ms);
             }
         }
     }
diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs
index e2fdd7b..62828a8 100644
--- a/src/node/lifecycle/mod.rs
+++ b/src/node/lifecycle/mod.rs
@@ -607,7 +607,7 @@ impl Node {
                 self.links.remove(&link_id);
                 self.addr_to_link
                     .remove(&(transport_id, remote_addr.clone()));
-                self.peer_machines.remove(&link_id);
+                self.remove_peer_machine(link_id);
                 return Err(NodeError::IndexAllocationFailed(e.to_string()));
             }
         };
@@ -623,7 +623,7 @@ impl Node {
                     self.links.remove(&link_id);
                     self.addr_to_link
                         .remove(&(transport_id, remote_addr.clone()));
-                    self.peer_machines.remove(&link_id);
+                    self.remove_peer_machine(link_id);
                     return Err(NodeError::HandshakeFailed(e.to_string()));
                 }
             };
@@ -1393,7 +1393,7 @@ impl Node {
                             );
                             // Clean up link and dial-time machine on handshake failure
                             self.remove_link(&pending.link_id);
-                            self.peer_machines.remove(&pending.link_id);
+                            self.remove_peer_machine(pending.link_id);
                         } else {
                             // Drive the dial-persisted machine: `Connecting` →
                             // `on_transport_connected` → `start_outbound_handshake`,
@@ -1460,7 +1460,7 @@ impl Node {
                 self.remove_link(&pending.link_id);
                 self.links.remove(&pending.link_id);
                 if let Some(id) = &pending.peer_identity {
-                    self.peer_machines.remove(&pending.link_id);
+                    self.remove_peer_machine(pending.link_id);
                     self.note_handshake_timeout(*id.node_addr(), Self::now_ms());
                 }
             }
diff --git a/src/node/mod.rs b/src/node/mod.rs
index 868c074..b22178e 100644
--- a/src/node/mod.rs
+++ b/src/node/mod.rs
@@ -50,7 +50,7 @@ use self::reloadable::Reloadable;
 pub(crate) const REKEY_JITTER_SECS: i64 = 15;
 use crate::cache::CoordCache;
 use crate::node::session::SessionEntry;
-use crate::peer::machine::PeerMachine;
+use crate::peer::machine::{PeerMachine, TimerKind};
 use crate::peer::{ActivePeer, PeerConnection};
 use crate::proto::bloom::{BloomFilter, BloomState};
 use crate::proto::fmp::Fmp;
@@ -380,6 +380,15 @@ pub struct Node {
     #[allow(dead_code)]
     peer_machines: HashMap<LinkId, PeerMachine>,
 
+    /// Per-peer timer store, keyed by `LinkId` then `TimerKind`, holding each
+    /// armed timer's absolute deadline (ms). The sans-IO time-as-input backing
+    /// for `PeerEvent::Timeout`: populated/cleared by the machine's
+    /// `SetTimer`/`CancelTimer` actions (`dataplane/peer_actions.rs`) and dropped
+    /// alongside the machine through the `remove_peer_machine` choke-point. At
+    /// this rung it is a SHADOW of the legacy tick timers — written but not yet
+    /// read by any driver (the handshake-kind fold wires the reader).
+    peer_timers: HashMap<LinkId, HashMap<TimerKind, u64>>,
+
     // === Peers (Active Phase) ===
     /// Authenticated peers.
     /// Indexed by NodeAddr (verified identity).
@@ -653,6 +662,7 @@ impl Node {
             child_exit_rx: None,
             connections: HashMap::new(),
             peer_machines: HashMap::new(),
+            peer_timers: HashMap::new(),
             peers: HashMap::new(),
             sessions: HashMap::new(),
             identity_cache: HashMap::new(),
@@ -803,6 +813,7 @@ impl Node {
             child_exit_rx: None,
             connections: HashMap::new(),
             peer_machines: HashMap::new(),
+            peer_timers: HashMap::new(),
             peers: HashMap::new(),
             sessions: HashMap::new(),
             identity_cache: HashMap::new(),
@@ -2013,7 +2024,7 @@ impl Node {
                 handshake_state: format!("{}", conn.handshake_state()),
                 started_at_ms: conn.started_at(),
                 last_activity_ms: conn.last_activity(),
-                resend_count: conn.resend_count(),
+                resend_count: self.connection_resend_count(conn.link_id()),
                 expected_peer: conn.expected_identity().map(|id| id.npub()),
             })
             .collect();
@@ -2287,6 +2298,26 @@ impl Node {
         }
     }
 
+    /// Single choke-point for dropping a per-peer control machine. Also drops the
+    /// machine's timer store so no armed `SetTimer` outlives it (the store and the
+    /// machine share the `LinkId` lifetime). Every teardown path routes machine
+    /// removal here rather than calling `peer_machines.remove` directly.
+    pub(in crate::node) fn remove_peer_machine(&mut self, link: LinkId) {
+        self.peer_machines.remove(&link);
+        self.peer_timers.remove(&link);
+    }
+
+    /// Operator-visible msg1 resend count for a pending handshake `link`, read
+    /// from the per-peer machine (the counter's home once the resend drive moved
+    /// off the shell connection). Machine-less connections (inbound legs that
+    /// never resend, and test-created connections) report 0, matching what the
+    /// shell connection reported before the counter moved.
+    pub(crate) fn connection_resend_count(&self, link: LinkId) -> u32 {
+        self.peer_machines
+            .get(&link)
+            .map_or(0, |machine| machine.resend_count())
+    }
+
     pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
         if !self
             .supervisor
diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs
index 593ee5a..b504673 100644
--- a/src/node/tests/handshake.rs
+++ b/src/node/tests/handshake.rs
@@ -899,28 +899,125 @@ async fn test_resend_scheduling() {
     );
     node.links.insert(link_id, link);
     node.addr_to_link
-        .insert((transport_id, remote_addr), link_id);
+        .insert((transport_id, remote_addr.clone()), link_id);
     node.pending_outbound
         .insert((transport_id, our_index.as_u32()), link_id);
     node.connections.insert(link_id, conn);
 
-    // Before resend time: nothing should happen (no transport = can't send,
-    // but the filter should exclude it because now < next_resend_at)
-    node.resend_pending_handshakes(now_ms + 500).await;
-    let conn = node.connections.get(&link_id).unwrap();
-    assert_eq!(conn.resend_count(), 0, "No resend before scheduled time");
+    // The msg1-resend counter and its due timer live on the per-peer machine.
+    // Dial it to `SentMsg1` (connectionless: no connect step) and arm its
+    // retransmit timer at now + 1000ms, mirroring what a real dial arms.
+    let mut machine =
+        crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, now_ms);
+    let _ = machine.step(
+        crate::peer::machine::PeerEvent::Dial {
+            transport_id,
+            remote_addr: remote_addr.clone(),
+            peer_identity,
+            connection_oriented: false,
+        },
+        now_ms,
+        &mut node.index_allocator,
+    );
+    node.peer_machines.insert(link_id, machine);
+    node.peer_timers.entry(link_id).or_default().insert(
+        crate::peer::machine::TimerKind::HandshakeRetransmit,
+        now_ms + 1000,
+    );
 
-    // At resend time: would resend if transport existed. Without transport,
-    // the send fails silently and resend_count stays at 0.
-    // This tests the filtering logic — the connection IS a candidate.
-    node.resend_pending_handshakes(now_ms + 1000).await;
-    // No transport registered, so send fails — count stays 0.
-    // That's the expected behavior (transport absence is a transient condition).
-    let conn = node.connections.get(&link_id).unwrap();
+    // Before the scheduled time the timer isn't due, so nothing fires.
+    node.drive_peer_timers(now_ms + 500).await;
     assert_eq!(
-        conn.resend_count(),
+        node.connection_resend_count(link_id),
         0,
-        "No transport means no resend recorded"
+        "No resend before scheduled time"
+    );
+
+    // At the scheduled time the timer is due, but no transport is registered so
+    // the send fails. Record-on-success: the count does NOT advance (and the
+    // connection is not marked failed) — a failed resend just retries next tick.
+    node.drive_peer_timers(now_ms + 1000).await;
+    assert_eq!(
+        node.connection_resend_count(link_id),
+        0,
+        "Failed send records no resend"
+    );
+}
+
+/// Test that the timer driver reaps an outbound leg whose machine
+/// `HandshakeTimeout` timer has come due (the timeout fold). The reap re-checks
+/// the shell `is_timed_out` predicate, then tears the connection down exactly as
+/// the old `check_timeouts` Teardown path did.
+#[tokio::test]
+async fn test_handshake_timeout_drive() {
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+    let peer_identity = make_peer_identity();
+    let remote_addr = TransportAddr::from_string("10.0.0.2:2121");
+
+    let dial_ms = 1000u64;
+    let link_id = node.allocate_link_id();
+    let mut conn = PeerConnection::outbound(link_id, peer_identity, dial_ms);
+    let our_index = node.index_allocator.allocate().unwrap();
+    let our_keypair = node.identity().keypair();
+    let _ = conn
+        .start_handshake(our_keypair, node.startup_epoch(), dial_ms)
+        .unwrap();
+    conn.set_our_index(our_index);
+    conn.set_transport_id(transport_id);
+    conn.set_source_addr(remote_addr.clone());
+
+    let link = Link::connectionless(
+        link_id,
+        transport_id,
+        remote_addr.clone(),
+        LinkDirection::Outbound,
+        Duration::from_millis(100),
+    );
+    node.links.insert(link_id, link);
+    node.addr_to_link
+        .insert((transport_id, remote_addr.clone()), link_id);
+    node.connections.insert(link_id, conn);
+    node.pending_outbound
+        .insert((transport_id, our_index.as_u32()), link_id);
+
+    // Machine in SentMsg1 with a HandshakeTimeout timer armed at dial + 30s.
+    let mut machine =
+        crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, dial_ms);
+    let _ = machine.step(
+        crate::peer::machine::PeerEvent::Dial {
+            transport_id,
+            remote_addr: remote_addr.clone(),
+            peer_identity,
+            connection_oriented: false,
+        },
+        dial_ms,
+        &mut node.index_allocator,
+    );
+    node.peer_machines.insert(link_id, machine);
+    node.peer_timers.entry(link_id).or_default().insert(
+        crate::peer::machine::TimerKind::HandshakeTimeout,
+        dial_ms + 30_000,
+    );
+
+    assert_eq!(node.connection_count(), 1);
+
+    // Well past dial + 30s: the timer is due and the leg is idle-timed-out.
+    node.drive_peer_timers(dial_ms + 100_000).await;
+
+    assert_eq!(
+        node.connection_count(),
+        0,
+        "Timed-out leg reaped by the timer drive"
+    );
+    assert_eq!(node.index_allocator.count(), 0, "Session index freed");
+    assert!(
+        !node.peer_machines.contains_key(&link_id),
+        "Control machine dropped with the reaped connection"
+    );
+    assert!(
+        !node.peer_timers.contains_key(&link_id),
+        "Timer store dropped with the reaped connection"
     );
 }
 
diff --git a/src/peer/machine.rs b/src/peer/machine.rs
index 6fa73d8..c46bb12 100644
--- a/src/peer/machine.rs
+++ b/src/peer/machine.rs
@@ -194,7 +194,14 @@ pub(crate) enum FailReason {
 }
 
 /// A timer the machine schedules on the driver's quantized tick.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+///
+/// `Hash` lets it key the driver's per-peer timer store; `Ord` lets the driver
+/// collect due kinds deterministically. Note the driver must fire
+/// `HandshakeTimeout` before `HandshakeRetransmit` on a same-tick coincidence
+/// (a reaped leg must not be resent), which is the reverse of this declaration
+/// order — the driver orders explicitly rather than relying on the derived
+/// ascending `Ord`.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub(crate) enum TimerKind {
     HandshakeRetransmit,
     HandshakeTimeout,
@@ -536,7 +543,37 @@ impl PeerMachine {
         self.our_index
     }
 
-    /// The crystallized node address, if known.
+    /// The msg1 resend count for this outbound handshake leg. The per-peer
+    /// machine is the home for this counter — the timer driver advances it via
+    /// [`record_resend`](Self::record_resend) on each successful resend, and the
+    /// control-socket connection snapshot reads it here so the operator-visible
+    /// count follows the machine rather than the (now inert) shell connection.
+    pub(crate) fn resend_count(&self) -> u32 {
+        self.conn.resend_count()
+    }
+
+    /// Record a successful msg1 resend: advance the count and store the next
+    /// backoff deadline. The driver calls this only after the resend actually
+    /// went out (record-on-success — a failed send neither advances the count
+    /// nor reschedules), matching the pre-fold shell semantics.
+    pub(crate) fn record_resend(&mut self, next_resend_at_ms: u64) {
+        self.conn.record_resend(next_resend_at_ms);
+    }
+
+    /// Whether this is an outbound leg parked at `SentMsg1` — the only state in
+    /// which a msg1 resend is due. Mirrors `on_handshake_retransmit`'s guard so
+    /// the shell timer driver can gate without reaching into machine state.
+    pub(crate) fn is_handshaking_sent_msg1(&self) -> bool {
+        matches!(
+            self.state,
+            PeerState::Handshaking {
+                phase: HandshakePhase::SentMsg1,
+                ..
+            }
+        )
+    }
+
+    /// The crystallized node address, if identity is known.
     fn addr(&self) -> Option<NodeAddr> {
         self.node_addr
     }
@@ -806,7 +843,19 @@ impl PeerMachine {
         self.conn.set_their_index(their_index);
         let addr = self.addr().unwrap_or_else(zero_addr);
         self.state = PeerState::Established { addr };
-        vec![PeerAction::PromoteToActive { link: self.link }]
+        // The machine survives promotion (it becomes the active peer's control
+        // machine), so cancel the outbound handshake timers here or they would
+        // linger in the driver's store. A late fire would no-op against the
+        // non-`Handshaking` state, but leaving them armed is a timer leak.
+        vec![
+            PeerAction::CancelTimer {
+                kind: TimerKind::HandshakeRetransmit,
+            },
+            PeerAction::CancelTimer {
+                kind: TimerKind::HandshakeTimeout,
+            },
+            PeerAction::PromoteToActive { link: self.link },
+        ]
     }
 
     /// XX inbound cross-connection at msg3. The Noise session swap and the
@@ -1426,10 +1475,50 @@ mod tests {
 
         let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc);
 
-        assert_eq!(actions, vec![PeerAction::PromoteToActive { link }]);
+        assert_eq!(
+            actions,
+            vec![
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeRetransmit
+                },
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeTimeout
+                },
+                PeerAction::PromoteToActive { link },
+            ]
+        );
         assert_eq!(m.state(), PeerState::Established { addr });
     }
 
+    // ---- Test: outbound promote at msg2 cancels handshake timers -----------
+    // The promoted machine survives as the active peer's control machine, so
+    // its outbound handshake retransmit + timeout timers must be cancelled at
+    // promotion or they linger in the driver's store. The promote must emit the
+    // two CancelTimer actions ahead of PromoteToActive, in that exact order.
+    #[test]
+    fn outbound_msg2_promote_cancels_handshake_timers() {
+        let mut alloc = IndexAllocator::new();
+        let id = peer_identity();
+        let link = LinkId::new(4);
+        let mut m = PeerMachine::new_outbound(link, id, 0);
+        let their_index = SessionIndex::new(0x88);
+
+        let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc);
+
+        assert_eq!(
+            actions,
+            vec![
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeRetransmit
+                },
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeTimeout
+                },
+                PeerAction::PromoteToActive { link },
+            ]
+        );
+    }
+
     // ---- Contract: actions are runtime-agnostic data ----------------------
     //
     // The machine's emitted actions are the message contract between the sync
@@ -2045,9 +2134,17 @@ mod tests {
         );
         assert_eq!(
             promote,
-            vec![PeerAction::PromoteToActive {
-                link: LinkId::new(1)
-            }]
+            vec![
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeRetransmit
+                },
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeTimeout
+                },
+                PeerAction::PromoteToActive {
+                    link: LinkId::new(1)
+                }
+            ]
         );
         assert_eq!(
             m.our_index(),
@@ -2142,9 +2239,17 @@ mod tests {
         );
         assert_eq!(
             promote,
-            vec![PeerAction::PromoteToActive {
-                link: LinkId::new(1)
-            }]
+            vec![
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeRetransmit
+                },
+                PeerAction::CancelTimer {
+                    kind: TimerKind::HandshakeTimeout
+                },
+                PeerAction::PromoteToActive {
+                    link: LinkId::new(1)
+                }
+            ]
         );
         assert_eq!(m.our_index(), None);
     }
diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs
index 2628a3e..3043d20 100644
--- a/src/proto/fmp/core.rs
+++ b/src/proto/fmp/core.rs
@@ -328,13 +328,6 @@ pub(crate) trait LifecycleView {
     /// timeout/failed predicate; the core decides retry-then-teardown.
     fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec<ConnSnapshot>;
 
-    /// Snapshot every outbound connection whose stored msg1 is due for a
-    /// resend as of `now_ms` and still under `max_resends`. The shell resolves
-    /// the "outbound, in `SentMsg1`, has stored msg1, under budget, past the
-    /// scheduled time" predicate and copies the opaque msg1 bytes; the core
-    /// computes the backoff schedule.
-    fn resend_candidates(&self, now_ms: u64, max_resends: u32) -> Vec<ConnSnapshot>;
-
     /// Snapshot every active peer with a session that is healthy, pre-computing
     /// its rekey-relevant ages and timer predicates (see [`PeerSnapshot`]). The
     /// shell resolves every clock read here; the core applies the thresholds.