Extract the nostr rendezvous decision logic into sans-IO state machines

Carve the nostr rendezvous engine's decision logic out of the async
NostrRendezvous driver into synchronous, clock-injected cores, following
the failure_state.rs pattern: sync state behind std::sync::Mutex, time
passed in as now_ms, no .await and no I/O in the core. The driver performs
all relay/socket I/O and executes the returned plans.

- AdvertMachine (src/nostr/advert.rs): advert publish/cache/fetch/prune
  decision logic; the driver executes a returned PublishPlan.
- TraversalMachine (src/nostr/traversal_machine.rs): the engine-scoped
  cross-session decision state -- in-flight-initiator dedup, the dual-init
  responder suppression election, and the replay/seen-session cache.
- classify_punch_packet (src/nostr/traversal.rs): a pure classifier for
  the punch recv loop's packet branch table.

The driver keeps all I/O, the NAT punch send cadence, the offer-slot
admission semaphore, and the pending-answer oneshot routing. Behavior-
neutral: no wire, config-key, or metric change. Adds unit coverage for
the advert plans, election ordering, replay eviction, initiator dedup,
and punch classification.

src/nostr/advert.rs

@@ -0,0 +1,505 @@
+//! Synchronous decision core for the Nostr overlay-advert lifecycle.
+//!
+//! `AdvertMachine` owns the advert-related state that previously lived
+//! directly on `NostrRendezvous` — the peer advert cache, the local
+//! advert we publish, and the id of our most recently published advert
+//! event — and hosts the *decision* logic for publishing, caching,
+//! fetching, and pruning adverts.
+//!
+//! Following the sans-IO shape used by `failure_state`, every method here
+//! is synchronous, performs no network I/O and no `.await`, holds its
+//! state behind `std::sync::Mutex`, and takes the current time as an
+//! explicit `now_ms: u64` input rather than reading a clock. The async
+//! driver on `NostrRendezvous` reads the clock at the call site, invokes
+//! these methods, and performs the actual relay I/O (`send_event_to`,
+//! `fetch_events_from`, gift-wrap crypto), event signing, NIP-09 deletes,
+//! and `Notify` wakeups described by the returned decisions.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+use nostr::prelude::{Event, EventId};
+
+use super::runtime::{NostrRendezvous, endpoint_advert_is_publicly_usable};
+use super::types::{
+    ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapError, CachedOverlayAdvert, OverlayAdvert,
+    OverlayEndpointAdvert,
+};
+
+/// What the async driver should do to satisfy a publish request. Returned
+/// by [`AdvertMachine::plan_publish`]; the machine performs the pure
+/// decision (which advert body, or a delete, or nothing) and the driver
+/// executes the corresponding relay I/O.
+#[derive(Debug, Clone)]
+pub(super) enum PublishPlan {
+    /// Nothing to publish (advertising disabled with no prior event, no
+    /// local advert yet, or the advert has no publicly usable endpoints).
+    Nothing,
+    /// Advertising is disabled but a prior advert event exists; the driver
+    /// should emit a NIP-09 delete for `EventId` then call
+    /// [`AdvertMachine::clear_event_id`].
+    Delete(EventId),
+    /// Publish this fully-prepared advert body. The driver builds the
+    /// tags/expiration, signs, sends, then records the new event id via
+    /// [`AdvertMachine::set_event_id`].
+    Publish(OverlayAdvert),
+}
+
+pub(super) struct AdvertMachine {
+    /// Our own npub. Used to avoid logging/attributing self-authored
+    /// adverts as peer discoveries.
+    npub: String,
+    /// Whether this node advertises at all (`config.advertise`).
+    advertise: bool,
+    /// Grace-extended max age for a cached advert, in ms
+    /// (`advert_ttl_secs * 1000 * stale-grace-multiplier`).
+    advert_max_age_ms: u64,
+    /// Size cap for the peer advert cache.
+    cache_max_entries: usize,
+    /// Peer advert cache keyed by author npub.
+    cache: Mutex<HashMap<String, CachedOverlayAdvert>>,
+    /// The advert body we currently want to publish, if any.
+    local_advert: Mutex<Option<OverlayAdvert>>,
+    /// Id of our most recently published advert event (for NIP-09 delete
+    /// on withdrawal).
+    current_event_id: Mutex<Option<EventId>>,
+}
+
+impl AdvertMachine {
+    pub(super) fn new(
+        npub: String,
+        advertise: bool,
+        advert_max_age_ms: u64,
+        cache_max_entries: usize,
+    ) -> Self {
+        Self {
+            npub,
+            advertise,
+            advert_max_age_ms,
+            cache_max_entries,
+            cache: Mutex::new(HashMap::new()),
+            local_advert: Mutex::new(None),
+            current_event_id: Mutex::new(None),
+        }
+    }
+
+    // --- validity (time-injected) --------------------------------------
+
+    /// Compute the validity horizon of an advert event, or `None` if it is
+    /// already stale. Thin time-injected wrapper over the pure
+    /// `compute_advert_valid_until_ms`.
+    pub(super) fn event_valid_until_ms(&self, event: &Event, now_ms: u64) -> Option<u64> {
+        NostrRendezvous::compute_advert_valid_until_ms(event, self.advert_max_age_ms, now_ms)
+    }
+
+    // --- cache: prune / observe / fetch --------------------------------
+
+    /// TTL + size-cap eviction. Drops entries past their validity horizon,
+    /// then evicts the oldest (by `valid_until_ms`) beyond the cap.
+    ///
+    /// Returns `Some((evicted, retained))` when a size-cap eviction
+    /// occurred so the driver can log it; `None` otherwise.
+    pub(super) fn prune(&self, now_ms: u64) -> Option<(usize, usize)> {
+        let mut cache = self.lock_cache();
+        cache.retain(|_, entry| entry.valid_until_ms > now_ms);
+        if cache.len() <= self.cache_max_entries {
+            return None;
+        }
+
+        let mut oldest = cache
+            .iter()
+            .map(|(npub, entry)| (npub.clone(), entry.valid_until_ms))
+            .collect::<Vec<_>>();
+        oldest.sort_by_key(|(_, ts)| *ts);
+        let overflow = cache.len().saturating_sub(self.cache_max_entries);
+        for (npub, _) in oldest.into_iter().take(overflow) {
+            cache.remove(&npub);
+        }
+        Some((overflow, cache.len()))
+    }
+
+    /// Observe an advert event received on the notify loop. Replaces the
+    /// cached entry iff its `created_at` is newer-or-equal to the cached
+    /// one (or none is cached).
+    ///
+    /// Returns `true` when the caller should log a "peer cached" line —
+    /// i.e. the entry was (re)cached *and* it is not our own advert.
+    pub(super) fn observe_advert(
+        &self,
+        author_npub: &str,
+        advert: OverlayAdvert,
+        created_at: u64,
+        valid_until_ms: u64,
+    ) -> bool {
+        let mut cache = self.lock_cache();
+        let should_replace = cache
+            .get(author_npub)
+            .map(|existing| existing.created_at <= created_at)
+            .unwrap_or(true);
+        if !should_replace {
+            return false;
+        }
+        let is_peer = author_npub != self.npub;
+        cache.insert(
+            author_npub.to_string(),
+            CachedOverlayAdvert {
+                author_npub: author_npub.to_string(),
+                advert,
+                created_at,
+                valid_until_ms,
+            },
+        );
+        is_peer
+    }
+
+    /// Cache-hit lookup for the fetch path: return the cached advert body
+    /// if present, `None` if the driver must fetch from relays.
+    pub(super) fn cached_advert(&self, peer_npub: &str) -> Option<OverlayAdvert> {
+        self.lock_cache()
+            .get(peer_npub)
+            .map(|cached| cached.advert.clone())
+    }
+
+    /// The `created_at` of a cached advert, if any. Used by the stale-check
+    /// refetch path to decide whether a relay result is newer.
+    pub(super) fn cached_created_at(&self, peer_npub: &str) -> Option<u64> {
+        self.lock_cache()
+            .get(peer_npub)
+            .map(|cached| cached.created_at)
+    }
+
+    /// Insert a freshly-fetched advert into the cache (fetch-miss path and
+    /// stale-check refresh).
+    pub(super) fn insert_fetched(&self, peer_npub: &str, cached: CachedOverlayAdvert) {
+        self.lock_cache().insert(peer_npub.to_string(), cached);
+    }
+
+    /// Remove a peer's cached advert (stale-check eviction).
+    pub(super) fn remove(&self, peer_npub: &str) {
+        self.lock_cache().remove(peer_npub);
+    }
+
+    /// Validity-filtered snapshot of cacheable peers for open discovery:
+    /// entries authored by someone other than us and still valid at
+    /// `now_ms`.
+    pub(super) fn open_discovery_candidates(
+        &self,
+        max: usize,
+        now_ms: u64,
+    ) -> Vec<(String, Vec<OverlayEndpointAdvert>, u64)> {
+        let cache = self.lock_cache();
+        cache
+            .values()
+            .filter(|entry| entry.author_npub != self.npub)
+            .filter(|entry| entry.valid_until_ms > now_ms)
+            .map(|entry| {
+                (
+                    entry.author_npub.clone(),
+                    entry.advert.endpoints.clone(),
+                    entry.created_at,
+                )
+            })
+            .take(max)
+            .collect()
+    }
+
+    // --- local advert / publish ----------------------------------------
+
+    /// Set the local advert we want to publish. Returns `true` when the
+    /// value changed (so the driver should request a republish).
+    pub(super) fn set_local_advert(&self, advert: Option<OverlayAdvert>) -> bool {
+        let mut slot = self.lock_local();
+        if *slot == advert {
+            false
+        } else {
+            *slot = advert;
+            true
+        }
+    }
+
+    /// Build the publish decision: which advert body to publish, a delete
+    /// to emit, or nothing. Pure logic — the driver performs the relay I/O
+    /// and event signing.
+    pub(super) fn plan_publish(&self) -> Result<PublishPlan, BootstrapError> {
+        let previous_event_id = *self.lock_event_id();
+        if !self.advertise {
+            return Ok(match previous_event_id {
+                Some(event_id) => PublishPlan::Delete(event_id),
+                None => PublishPlan::Nothing,
+            });
+        }
+
+        let mut advert = match self.lock_local().clone() {
+            Some(advert) => advert,
+            // Transient absence (e.g., a single tick during startup where
+            // build_overlay_advert briefly returns None). Don't proactively
+            // emit a NIP-09 delete: the next publish supersedes the old
+            // event via parameterized-replaceable semantics, and the NIP-40
+            // expiration tag bounds the worst case if we never re-publish.
+            None => return Ok(PublishPlan::Nothing),
+        };
+
+        advert.identifier = ADVERT_IDENTIFIER.to_string();
+        advert.version = ADVERT_VERSION;
+        advert.endpoints.retain(endpoint_advert_is_publicly_usable);
+        // Defensive: build_overlay_advert returns None on empty endpoints,
+        // so this is only reachable from non-lifecycle callers.
+        if advert.endpoints.is_empty() {
+            return Ok(PublishPlan::Nothing);
+        }
+
+        if advert.has_udp_nat_endpoint() {
+            if advert
+                .signal_relays
+                .as_ref()
+                .is_none_or(|relays| relays.is_empty())
+            {
+                return Err(BootstrapError::InvalidAdvert(
+                    "udp:nat endpoint requires non-empty signalRelays".to_string(),
+                ));
+            }
+            if advert
+                .stun_servers
+                .as_ref()
+                .is_none_or(|servers| servers.is_empty())
+            {
+                return Err(BootstrapError::InvalidAdvert(
+                    "udp:nat endpoint requires non-empty stunServers".to_string(),
+                ));
+            }
+        } else {
+            advert.signal_relays = None;
+            advert.stun_servers = None;
+        }
+
+        Ok(PublishPlan::Publish(advert))
+    }
+
+    // --- current advert event id (NIP-09 delete-on-withdraw) -----------
+
+    /// Record the id of a just-published advert event.
+    pub(super) fn set_event_id(&self, event_id: EventId) {
+        *self.lock_event_id() = Some(event_id);
+    }
+
+    /// Clear the recorded advert event id (after emitting a delete).
+    pub(super) fn clear_event_id(&self) {
+        *self.lock_event_id() = None;
+    }
+
+    /// Take and clear the recorded advert event id (shutdown path).
+    pub(super) fn take_event_id(&self) -> Option<EventId> {
+        self.lock_event_id().take()
+    }
+
+    // --- lock helpers ---------------------------------------------------
+
+    fn lock_cache(&self) -> std::sync::MutexGuard<'_, HashMap<String, CachedOverlayAdvert>> {
+        self.cache
+            .lock()
+            .expect("advert-machine cache mutex poisoned")
+    }
+
+    fn lock_local(&self) -> std::sync::MutexGuard<'_, Option<OverlayAdvert>> {
+        self.local_advert
+            .lock()
+            .expect("advert-machine local-advert mutex poisoned")
+    }
+
+    fn lock_event_id(&self) -> std::sync::MutexGuard<'_, Option<EventId>> {
+        self.current_event_id
+            .lock()
+            .expect("advert-machine event-id mutex poisoned")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::nostr::types::OverlayTransportKind;
+
+    fn ep(addr: &str) -> OverlayEndpointAdvert {
+        OverlayEndpointAdvert {
+            transport: OverlayTransportKind::Udp,
+            addr: addr.to_string(),
+        }
+    }
+
+    fn advert(endpoints: Vec<OverlayEndpointAdvert>) -> OverlayAdvert {
+        OverlayAdvert {
+            identifier: ADVERT_IDENTIFIER.to_string(),
+            version: ADVERT_VERSION,
+            endpoints,
+            signal_relays: None,
+            stun_servers: None,
+        }
+    }
+
+    fn cached(author: &str, created_at: u64, valid_until_ms: u64) -> CachedOverlayAdvert {
+        CachedOverlayAdvert {
+            author_npub: author.to_string(),
+            advert: advert(vec![ep("1.2.3.4:9000")]),
+            created_at,
+            valid_until_ms,
+        }
+    }
+
+    fn machine() -> AdvertMachine {
+        // npub=self, advertise=true, max-age huge, cap=3
+        AdvertMachine::new("npub1self".to_string(), true, 10_000_000, 3)
+    }
+
+    #[test]
+    fn observe_advert_replaces_only_when_newer_and_flags_peer() {
+        let m = machine();
+        // Fresh peer advert -> cached, log flagged (peer).
+        assert!(m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9000")]), 100, 5000));
+        // Older created_at -> not replaced, no log.
+        assert!(!m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9001")]), 50, 5000));
+        assert_eq!(m.cached_created_at("npub1peer"), Some(100));
+        // Newer created_at -> replaced, log flagged.
+        assert!(m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9002")]), 200, 5000));
+        assert_eq!(m.cached_created_at("npub1peer"), Some(200));
+    }
+
+    #[test]
+    fn observe_advert_self_author_caches_but_does_not_flag_log() {
+        let m = machine();
+        // Own advert is still cached (should_replace true) but must NOT be
+        // flagged as a peer-cached log line.
+        assert!(!m.observe_advert("npub1self", advert(vec![ep("1.2.3.4:9000")]), 100, 5000));
+        assert_eq!(m.cached_created_at("npub1self"), Some(100));
+    }
+
+    #[test]
+    fn prune_drops_expired_and_reports_no_eviction_under_cap() {
+        let m = machine();
+        m.insert_fetched("npub1a", cached("npub1a", 1, 1000));
+        m.insert_fetched("npub1b", cached("npub1b", 1, 3000));
+        // now=2000 -> npub1a expired, npub1b retained, under cap -> None.
+        assert_eq!(m.prune(2000), None);
+        assert_eq!(m.cached_created_at("npub1a"), None);
+        assert!(m.cached_created_at("npub1b").is_some());
+    }
+
+    #[test]
+    fn prune_size_cap_evicts_oldest_by_valid_until() {
+        let m = machine(); // cap = 3
+        // Four still-valid entries; oldest valid_until must be evicted.
+        m.insert_fetched("npub1a", cached("npub1a", 1, 1000));
+        m.insert_fetched("npub1b", cached("npub1b", 1, 2000));
+        m.insert_fetched("npub1c", cached("npub1c", 1, 3000));
+        m.insert_fetched("npub1d", cached("npub1d", 1, 4000));
+        let evicted = m.prune(500);
+        assert_eq!(evicted, Some((1, 3)));
+        // Oldest validity (npub1a) evicted; newest kept.
+        assert_eq!(m.cached_created_at("npub1a"), None);
+        assert!(m.cached_created_at("npub1d").is_some());
+    }
+
+    #[test]
+    fn open_discovery_candidates_filters_self_and_expired() {
+        let m = machine();
+        m.insert_fetched("npub1self", cached("npub1self", 1, 9000));
+        m.insert_fetched("npub1peer", cached("npub1peer", 1, 9000));
+        m.insert_fetched("npub1stale", cached("npub1stale", 1, 1000));
+        let out = m.open_discovery_candidates(10, 2000);
+        assert_eq!(out.len(), 1, "only the valid non-self peer survives");
+        assert_eq!(out[0].0, "npub1peer");
+    }
+
+    #[test]
+    fn open_discovery_candidates_respects_max() {
+        let m = machine();
+        for i in 0..5 {
+            let npub = format!("npub1p{i}");
+            m.insert_fetched(&npub, cached(&npub, 1, 9000));
+        }
+        assert_eq!(m.open_discovery_candidates(2, 1000).len(), 2);
+    }
+
+    #[test]
+    fn set_local_advert_detects_change() {
+        let m = machine();
+        let a = advert(vec![ep("1.2.3.4:9000")]);
+        assert!(m.set_local_advert(Some(a.clone())), "first set is a change");
+        assert!(
+            !m.set_local_advert(Some(a.clone())),
+            "identical set is no change"
+        );
+        assert!(m.set_local_advert(None), "clearing is a change");
+    }
+
+    #[test]
+    fn plan_publish_strips_relays_for_non_nat_advert() {
+        let m = machine();
+        let mut a = advert(vec![ep("1.2.3.4:9000")]);
+        a.signal_relays = Some(vec!["wss://relay".to_string()]);
+        a.stun_servers = Some(vec!["stun:host:3478".to_string()]);
+        m.set_local_advert(Some(a));
+        match m.plan_publish().expect("plan ok") {
+            PublishPlan::Publish(out) => {
+                assert!(
+                    out.signal_relays.is_none(),
+                    "non-nat advert strips signalRelays"
+                );
+                assert!(
+                    out.stun_servers.is_none(),
+                    "non-nat advert strips stunServers"
+                );
+                assert_eq!(out.identifier, ADVERT_IDENTIFIER);
+                assert_eq!(out.version, ADVERT_VERSION);
+            }
+            other => panic!("expected Publish, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn plan_publish_keeps_relays_for_nat_advert() {
+        let m = machine();
+        let mut a = advert(vec![ep("nat")]);
+        a.signal_relays = Some(vec!["wss://relay".to_string()]);
+        a.stun_servers = Some(vec!["stun:host:3478".to_string()]);
+        m.set_local_advert(Some(a));
+        match m.plan_publish().expect("plan ok") {
+            PublishPlan::Publish(out) => {
+                assert!(out.has_udp_nat_endpoint());
+                assert_eq!(out.signal_relays.as_deref().map(<[_]>::len), Some(1));
+                assert_eq!(out.stun_servers.as_deref().map(<[_]>::len), Some(1));
+            }
+            other => panic!("expected Publish, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn plan_publish_nat_without_relays_errors() {
+        let m = machine();
+        m.set_local_advert(Some(advert(vec![ep("nat")])));
+        assert!(matches!(
+            m.plan_publish(),
+            Err(BootstrapError::InvalidAdvert(_))
+        ));
+    }
+
+    #[test]
+    fn plan_publish_nothing_when_no_local_advert() {
+        let m = machine();
+        assert!(matches!(m.plan_publish(), Ok(PublishPlan::Nothing)));
+    }
+
+    #[test]
+    fn plan_publish_nothing_when_disabled_without_prior_event() {
+        // advertise=false, no prior event id -> Nothing.
+        let m = AdvertMachine::new("npub1self".to_string(), false, 10_000_000, 3);
+        m.set_local_advert(Some(advert(vec![ep("1.2.3.4:9000")])));
+        assert!(matches!(m.plan_publish(), Ok(PublishPlan::Nothing)));
+    }
+
+    #[test]
+    fn event_id_set_clear_take_roundtrip() {
+        let m = machine();
+        assert_eq!(m.take_event_id(), None);
+        m.clear_event_id();
+        assert_eq!(m.take_event_id(), None);
+    }
+}

src/nostr/mod.rs

@@ -1,3 +1,4 @@
+mod advert;
 mod driver;
 mod failure_state;
 mod handoff;
@@ -5,6 +6,7 @@ mod runtime;
 mod signal;
 mod stun;
 mod traversal;
+mod traversal_machine;
 mod types;
 
 #[cfg(test)]

src/nostr/runtime.rs

@@ -16,6 +16,7 @@ use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 use tracing::{debug, info, trace, warn};
 
+use super::advert::{AdvertMachine, PublishPlan};
 use super::failure_state::FailureState;
 use super::handoff::EstablishedTraversal;
 use super::signal::{
@@ -25,14 +26,15 @@ use super::signal::{
 };
 use super::stun::observe_traversal_addresses;
 use super::traversal::{nonce, now_ms, planned_remote_endpoints, run_punch_attempt};
+use super::traversal_machine::{OfferDisposition, SeenDecision, TraversalMachine};
 use super::types::{
     ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent,
     CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome,
     OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint, SIGNAL_KIND,
     TraversalAnswer, TraversalOffer,
 };
+use crate::PeerIdentity;
 use crate::config::{NostrRendezvousConfig, PeerConfig};
-use crate::{NodeAddr, PeerIdentity};
 
 const ADVERT_CACHE_STALE_GRACE_MULTIPLIER: u64 = 2;
 
@@ -51,42 +53,6 @@ fn short_id(id: &str) -> String {
     }
 }
 
-/// Decide whether an incoming-offer responder session should be suppressed
-/// in favour of our own already-running outbound initiator session.
-///
-/// Two peers that each have the other as `auto_connect` simultaneously run an
-/// initiator traversal *and* a responder traversal for the same peer, binding a
-/// separate UDP socket per session. Each node then emits two
-/// `BootstrapEvent::Established` events and `adopt_established_traversal` keeps
-/// only the first on a non-deterministic race; when the two nodes' independent
-/// races resolve to mismatched sessions, each side's Noise msg1 lands on a peer
-/// port the peer already stopped draining and both handshakes stall (root cause
-/// of ISSUE-2026-0031).
-///
-/// To collapse the four-socket dance to a single, guaranteed-matching socket
-/// pair, both nodes deterministically keep the session **initiated by the
-/// smaller `NodeAddr`** — reusing the project's existing NodeAddr tie-breaker
-/// convention (`cross_connection_winner`, the rekey dual-init resolution, and
-/// the dual-cross-init adopt path in `lifecycle.rs`).
-///
-/// This is evaluated on the responder path, where the session being handled is
-/// *peer-initiated*. It returns `true` (suppress this responder session) only
-/// when genuine duplication exists — i.e. we also have an in-flight outbound
-/// initiator for this same peer (`have_active_initiator`) — and our own
-/// initiator session is the preferred one (`our_addr < peer_addr`). When there
-/// is no co-active initiator (the asymmetric / one-sided `auto_connect` case,
-/// where only one session exists at all) it never suppresses, so connectivity
-/// is preserved. The `our_addr == peer_addr` case (self / loopback) and any
-/// caller that cannot derive a peer `NodeAddr` likewise fall through to "do not
-/// suppress".
-pub(super) fn suppress_responder_for_own_initiator(
-    our_addr: &NodeAddr,
-    peer_addr: &NodeAddr,
-    have_active_initiator: bool,
-) -> bool {
-    have_active_initiator && our_addr < peer_addr
-}
-
 fn endpoint_summary(endpoints: &[OverlayEndpointAdvert]) -> String {
     endpoints
         .iter()
@@ -117,7 +83,7 @@ fn is_unroutable_direct_advert_ip(ip: std::net::IpAddr) -> bool {
     }
 }
 
-fn endpoint_advert_is_publicly_usable(endpoint: &OverlayEndpointAdvert) -> bool {
+pub(super) fn endpoint_advert_is_publicly_usable(endpoint: &OverlayEndpointAdvert) -> bool {
     let addr = endpoint.addr.trim();
     if addr.is_empty() {
         return false;
@@ -183,12 +149,9 @@ pub struct NostrRendezvous {
     pubkey: PublicKey,
     npub: String,
     config: NostrRendezvousConfig,
-    advert_cache: RwLock<HashMap<String, CachedOverlayAdvert>>,
-    local_advert: RwLock<Option<OverlayAdvert>>,
-    current_advert_event_id: RwLock<Option<EventId>>,
+    advert: AdvertMachine,
+    traversal: TraversalMachine,
     pending_answers: Mutex<HashMap<String, oneshot::Sender<SignalEnvelope<TraversalAnswer>>>>,
-    active_initiators: Mutex<HashSet<String>>,
-    seen_sessions: Mutex<HashMap<String, u64>>,
     offer_slots: Arc<Semaphore>,
     event_tx: mpsc::UnboundedSender<BootstrapEvent>,
     event_rx: Mutex<mpsc::UnboundedReceiver<BootstrapEvent>>,
@@ -249,18 +212,25 @@ impl NostrRendezvous {
             config.failure_state_max_entries,
         );
 
+        let advert = AdvertMachine::new(
+            npub.clone(),
+            config.advertise,
+            config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER,
+            config.advert_cache_max_entries,
+        );
+        let traversal = TraversalMachine::new(
+            config.replay_window_secs * 1000,
+            config.seen_sessions_max_entries,
+        );
         let runtime = Arc::new(Self {
             client,
             keys,
             pubkey,
             npub,
             config,
-            advert_cache: RwLock::new(HashMap::new()),
-            local_advert: RwLock::new(None),
-            current_advert_event_id: RwLock::new(None),
+            advert,
+            traversal,
             pending_answers: Mutex::new(HashMap::new()),
-            active_initiators: Mutex::new(HashSet::new()),
-            seen_sessions: Mutex::new(HashMap::new()),
             offer_slots,
             event_tx,
             event_rx: Mutex::new(event_rx),
@@ -309,11 +279,8 @@ impl NostrRendezvous {
 
     pub async fn request_connect(self: &Arc<Self>, peer_config: PeerConfig) {
         let peer_npub = peer_config.npub.clone();
-        {
-            let mut active = self.active_initiators.lock().await;
-            if !active.insert(peer_npub.clone()) {
-                return;
-            }
+        if !self.traversal.begin_initiator(&peer_npub) {
+            return;
         }
 
         let runtime = Arc::clone(self);
@@ -326,7 +293,7 @@ impl NostrRendezvous {
                 },
             };
             let _ = runtime.event_tx.send(event);
-            runtime.active_initiators.lock().await.remove(&peer_npub);
+            runtime.traversal.end_initiator(&peer_npub);
         });
     }
 
@@ -507,12 +474,7 @@ impl NostrRendezvous {
         if self.config.advert_relays.is_empty() {
             return NostrRefetchOutcome::Skipped;
         }
-        let cached_created_at = self
-            .advert_cache
-            .read()
-            .await
-            .get(peer_npub)
-            .map(|c| c.created_at);
+        let cached_created_at = self.advert.cached_created_at(peer_npub);
 
         let events = match self
             .client
@@ -541,7 +503,7 @@ impl NostrRendezvous {
 
         let Some((relay_created_at, ev)) = newest else {
             // Absent on relays. Evict any stale cache entry.
-            self.advert_cache.write().await.remove(peer_npub);
+            self.advert.remove(peer_npub);
             self.failure_state.reset_streak_after_refresh(peer_npub);
             return NostrRefetchOutcome::Evicted;
         };
@@ -561,10 +523,7 @@ impl NostrRendezvous {
                     created_at: relay_created_at,
                     valid_until_ms,
                 };
-                self.advert_cache
-                    .write()
-                    .await
-                    .insert(peer_npub.to_string(), updated);
+                self.advert.insert_fetched(peer_npub, updated);
                 self.failure_state.reset_streak_after_refresh(peer_npub);
                 NostrRefetchOutcome::Refreshed
             }
@@ -584,19 +543,9 @@ impl NostrRendezvous {
         self: &Arc<Self>,
         advert: Option<OverlayAdvert>,
     ) -> Result<(), BootstrapError> {
-        let changed = {
-            let mut slot = self.local_advert.write().await;
-            if *slot == advert {
-                false
-            } else {
-                *slot = advert;
-                true
-            }
-        };
-        if !changed {
-            return Ok(());
+        if self.advert.set_local_advert(advert) {
+            self.request_publish_advert();
         }
-        self.request_publish_advert();
         Ok(())
     }
 
@@ -617,22 +566,8 @@ impl NostrRendezvous {
         &self,
         max: usize,
     ) -> Vec<(String, Vec<OverlayEndpointAdvert>, u64)> {
-        self.prune_advert_cache().await;
-        let now = now_ms();
-        let cache = self.advert_cache.read().await;
-        cache
-            .values()
-            .filter(|entry| entry.author_npub != self.npub)
-            .filter(|entry| entry.valid_until_ms > now)
-            .map(|entry| {
-                (
-                    entry.author_npub.clone(),
-                    entry.advert.endpoints.clone(),
-                    entry.created_at,
-                )
-            })
-            .take(max)
-            .collect()
+        self.prune_advert_cache();
+        self.advert.open_discovery_candidates(max, now_ms())
     }
 
     pub async fn shutdown(&self) -> Result<(), BootstrapError> {
@@ -655,7 +590,7 @@ impl NostrRendezvous {
         // permanent shutdown. An explicit retraction races with the next
         // daemon's republish on strict relays (e.g. Damus rate-limits the
         // burst, leaving the advert deleted and never restored).
-        let _ = self.current_advert_event_id.write().await.take();
+        let _ = self.advert.take_event_id();
 
         if let Some(handle) = self.notify_task.lock().await.take() {
             handle.abort();
@@ -701,32 +636,23 @@ impl NostrRendezvous {
                             && let Ok(advert) =
                                 Self::parse_overlay_advert_event(&event, &self.config.app)
                         {
-                            let mut cache = self.advert_cache.write().await;
-                            let should_replace = cache
-                                .get(&author_npub)
-                                .map(|existing| existing.created_at <= event.created_at.as_secs())
-                                .unwrap_or(true);
-                            if should_replace && author_npub != self.npub {
+                            let endpoints = endpoint_summary(&advert.endpoints);
+                            let created_at = event.created_at.as_secs();
+                            if self.advert.observe_advert(
+                                &author_npub,
+                                advert,
+                                created_at,
+                                valid_until_ms,
+                            ) {
                                 debug!(
                                     peer = %short_npub(&author_npub),
-                                    endpoints = %endpoint_summary(&advert.endpoints),
+                                    endpoints = %endpoints,
                                     event = %short_id(&event.id.to_string()),
                                     "advert: peer cached"
                                 );
                             }
-                            if should_replace {
-                                cache.insert(
-                                    author_npub.clone(),
-                                    CachedOverlayAdvert {
-                                        author_npub,
-                                        advert,
-                                        created_at: event.created_at.as_secs(),
-                                        valid_until_ms,
-                                    },
-                                );
-                            }
                         }
-                        self.prune_advert_cache().await;
+                        self.prune_advert_cache();
                         continue;
                     }
 
@@ -949,59 +875,17 @@ impl NostrRendezvous {
     }
 
     async fn publish_advert(&self) -> Result<(), BootstrapError> {
-        let previous_event_id = self.current_advert_event_id.read().await.to_owned();
-        if !self.config.advertise {
-            if let Some(event_id) = previous_event_id {
+        let advert = match self.advert.plan_publish()? {
+            PublishPlan::Nothing => return Ok(()),
+            PublishPlan::Delete(event_id) => {
                 self.publish_delete(&self.config.advert_relays, [event_id])
                     .await?;
-                *self.current_advert_event_id.write().await = None;
+                self.advert.clear_event_id();
+                return Ok(());
             }
-            return Ok(());
-        }
-
-        let mut advert = match self.local_advert.read().await.clone() {
-            Some(advert) => advert,
-            // Transient absence (e.g., a single tick during startup where
-            // build_overlay_advert briefly returns None). Don't proactively
-            // emit a NIP-09 delete: the next publish supersedes the old
-            // event via parameterized-replaceable semantics, and the NIP-40
-            // expiration tag bounds the worst case if we never re-publish.
-            None => return Ok(()),
+            PublishPlan::Publish(advert) => advert,
         };
 
-        advert.identifier = ADVERT_IDENTIFIER.to_string();
-        advert.version = ADVERT_VERSION;
-        advert.endpoints.retain(endpoint_advert_is_publicly_usable);
-        // Defensive: build_overlay_advert returns None on empty endpoints,
-        // so this is only reachable from non-lifecycle callers.
-        if advert.endpoints.is_empty() {
-            return Ok(());
-        }
-
-        if advert.has_udp_nat_endpoint() {
-            if advert
-                .signal_relays
-                .as_ref()
-                .is_none_or(|relays| relays.is_empty())
-            {
-                return Err(BootstrapError::InvalidAdvert(
-                    "udp:nat endpoint requires non-empty signalRelays".to_string(),
-                ));
-            }
-            if advert
-                .stun_servers
-                .as_ref()
-                .is_none_or(|servers| servers.is_empty())
-            {
-                return Err(BootstrapError::InvalidAdvert(
-                    "udp:nat endpoint requires non-empty stunServers".to_string(),
-                ));
-            }
-        } else {
-            advert.signal_relays = None;
-            advert.stun_servers = None;
-        }
-
         let expires_at = now_ms() + self.config.advert_ttl_secs * 1000;
         let tags = vec![
             Tag::identifier(ADVERT_IDENTIFIER.to_string()),
@@ -1031,7 +915,7 @@ impl NostrRendezvous {
         // NIP-09 delete here is redundant and races with the replacement
         // publish, which strict relays (e.g. Damus) honor by removing the
         // new advert too.
-        *self.current_advert_event_id.write().await = Some(event.id);
+        self.advert.set_event_id(event.id);
         Ok(())
     }
 
@@ -1270,17 +1154,17 @@ impl NostrRendezvous {
         // single matching socket pair survives on both sides. Asymmetric /
         // one-sided `auto_connect` (no co-active initiator) is never suppressed,
         // preserving connectivity. See `suppress_responder_for_own_initiator`.
-        if self.active_initiators.lock().await.contains(&sender_npub) {
-            match (
-                PeerIdentity::from_npub(&self.npub),
-                PeerIdentity::from_npub(&sender_npub),
-            ) {
-                (Ok(ours), Ok(theirs)) => {
-                    if suppress_responder_for_own_initiator(
-                        ours.node_addr(),
-                        theirs.node_addr(),
-                        true,
-                    ) {
+        match (
+            PeerIdentity::from_npub(&self.npub),
+            PeerIdentity::from_npub(&sender_npub),
+        ) {
+            (Ok(ours), Ok(theirs)) => {
+                match self.traversal.classify_incoming_offer(
+                    &sender_npub,
+                    ours.node_addr(),
+                    theirs.node_addr(),
+                ) {
+                    OfferDisposition::Suppress => {
                         debug!(
                             peer = %peer_short,
                             session = %short_id(&offer.session_id),
@@ -1288,20 +1172,38 @@ impl NostrRendezvous {
                         );
                         return Ok(());
                     }
+                    OfferDisposition::Proceed => {}
                 }
-                _ => {
-                    // Could not derive a NodeAddr for one side; fall through and
-                    // answer rather than risk suppressing the only session.
-                    trace!(
-                        peer = %peer_short,
-                        "traversal: could not derive NodeAddr for dedup, answering offer"
+            }
+            _ => {
+                // Could not derive a NodeAddr for one side; fall through and
+                // answer rather than risk suppressing the only session.
+                trace!(
+                    peer = %peer_short,
+                    "traversal: could not derive NodeAddr for dedup, answering offer"
+                );
+            }
+        }
+
+        match self
+            .traversal
+            .note_session_seen(&offer.session_id, now_ms())
+        {
+            SeenDecision::Replay => {
+                return Err(BootstrapError::Replay(offer.session_id.clone()));
+            }
+            SeenDecision::Fresh { evicted } => {
+                if let Some((evicted, retained)) = evicted {
+                    debug!(
+                        evicted = evicted,
+                        retained = retained,
+                        cap = self.config.seen_sessions_max_entries,
+                        "seen-sessions cache overflow; evicted oldest entries"
                     );
                 }
             }
         }
 
-        self.mark_session_seen(&offer.session_id).await?;
-
         let base_socket = std::net::UdpSocket::bind(("0.0.0.0", 0))?;
         base_socket.set_nonblocking(true)?;
         let (reflexive_address, local_addresses, stun_server) = observe_traversal_addresses(
@@ -1396,15 +1298,15 @@ impl NostrRendezvous {
         peer_npub: &str,
         target_pubkey: PublicKey,
     ) -> Result<OverlayAdvert, BootstrapError> {
-        self.prune_advert_cache().await;
-        if let Some(cached) = self.advert_cache.read().await.get(peer_npub).cloned() {
+        self.prune_advert_cache();
+        if let Some(advert) = self.advert.cached_advert(peer_npub) {
             debug!(
                 peer = %short_npub(peer_npub),
                 source = "cache",
-                endpoints = %endpoint_summary(&cached.advert.endpoints),
+                endpoints = %endpoint_summary(&advert.endpoints),
                 "advert: resolved"
             );
-            return Ok(cached.advert);
+            return Ok(advert);
         }
 
         let events = self
@@ -1453,11 +1355,8 @@ impl NostrRendezvous {
             endpoints = %endpoint_summary(&cached.advert.endpoints),
             "advert: resolved"
         );
-        self.advert_cache
-            .write()
-            .await
-            .insert(peer_npub.to_string(), cached.clone());
-        self.prune_advert_cache().await;
+        self.advert.insert_fetched(peer_npub, cached.clone());
+        self.prune_advert_cache();
         Ok(cached.advert)
     }
 
@@ -1603,39 +1502,19 @@ impl NostrRendezvous {
         Ok(advert)
     }
 
-    async fn prune_advert_cache(&self) {
-        let now = now_ms();
-        let mut cache = self.advert_cache.write().await;
-        cache.retain(|_, entry| entry.valid_until_ms > now);
-        if cache.len() <= self.config.advert_cache_max_entries {
-            return;
+    fn prune_advert_cache(&self) {
+        if let Some((evicted, retained)) = self.advert.prune(now_ms()) {
+            debug!(
+                evicted,
+                retained,
+                cap = self.config.advert_cache_max_entries,
+                "advert cache overflow; evicted oldest entries"
+            );
         }
-
-        let mut oldest = cache
-            .iter()
-            .map(|(npub, entry)| (npub.clone(), entry.valid_until_ms))
-            .collect::<Vec<_>>();
-        oldest.sort_by_key(|(_, ts)| *ts);
-        let overflow = cache
-            .len()
-            .saturating_sub(self.config.advert_cache_max_entries);
-        for (npub, _) in oldest.into_iter().take(overflow) {
-            cache.remove(&npub);
-        }
-        debug!(
-            evicted = overflow,
-            retained = cache.len(),
-            cap = self.config.advert_cache_max_entries,
-            "advert cache overflow; evicted oldest entries"
-        );
-    }
-
-    fn advert_max_age_ms(&self) -> u64 {
-        self.config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER
     }
 
     fn event_valid_until_ms(&self, event: &Event) -> Option<u64> {
-        Self::compute_advert_valid_until_ms(event, self.advert_max_age_ms(), now_ms())
+        self.advert.event_valid_until_ms(event, now_ms())
     }
 
     pub(super) fn compute_advert_valid_until_ms(
@@ -1699,37 +1578,6 @@ impl NostrRendezvous {
             .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
         Ok(())
     }
-
-    async fn mark_session_seen(&self, session_id: &str) -> Result<(), BootstrapError> {
-        let now = now_ms();
-        let expiry = now + self.config.replay_window_secs * 1000;
-        let mut seen = self.seen_sessions.lock().await;
-        seen.retain(|_, expires_at| *expires_at > now);
-        if seen.contains_key(session_id) {
-            return Err(BootstrapError::Replay(session_id.to_string()));
-        }
-        seen.insert(session_id.to_string(), expiry);
-        if seen.len() > self.config.seen_sessions_max_entries {
-            let mut oldest = seen
-                .iter()
-                .map(|(session, expires_at)| (session.clone(), *expires_at))
-                .collect::<Vec<_>>();
-            oldest.sort_by_key(|(_, expires_at)| *expires_at);
-            let overflow = seen
-                .len()
-                .saturating_sub(self.config.seen_sessions_max_entries);
-            for (session, _) in oldest.into_iter().take(overflow) {
-                seen.remove(&session);
-            }
-            debug!(
-                evicted = overflow,
-                retained = seen.len(),
-                cap = self.config.seen_sessions_max_entries,
-                "seen-sessions cache overflow; evicted oldest entries"
-            );
-        }
-        Ok(())
-    }
 }
 
 #[cfg(test)]
@@ -1755,18 +1603,25 @@ impl NostrRendezvous {
             config.warn_log_interval_secs,
             config.failure_state_max_entries,
         );
+        let advert = AdvertMachine::new(
+            npub.clone(),
+            config.advertise,
+            config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER,
+            config.advert_cache_max_entries,
+        );
+        let traversal = TraversalMachine::new(
+            config.replay_window_secs * 1000,
+            config.seen_sessions_max_entries,
+        );
         Self {
             client,
             keys,
             pubkey,
             npub,
             config,
-            advert_cache: RwLock::new(HashMap::new()),
-            local_advert: RwLock::new(None),
-            current_advert_event_id: RwLock::new(None),
+            advert,
+            traversal,
             pending_answers: Mutex::new(HashMap::new()),
-            active_initiators: Mutex::new(HashSet::new()),
-            seen_sessions: Mutex::new(HashMap::new()),
             offer_slots,
             event_tx,
             event_rx: Mutex::new(event_rx),
@@ -1806,8 +1661,7 @@ impl NostrRendezvous {
     /// Insert a cached advert directly into the in-memory cache. Used by
     /// unit tests to set up consumer-side state without needing live relays.
     pub(crate) async fn insert_advert_for_test(&self, npub: String, advert: CachedOverlayAdvert) {
-        let mut cache = self.advert_cache.write().await;
-        cache.insert(npub, advert);
+        self.advert.insert_fetched(&npub, advert);
     }
 
     /// Queue a bootstrap event directly for lifecycle tests without live relays

src/nostr/tests.rs

@@ -1,6 +1,6 @@
 use nostr::prelude::{EventBuilder, Kind, Tag, Timestamp};
 
-use super::runtime::{NostrRendezvous, suppress_responder_for_own_initiator};
+use super::runtime::NostrRendezvous;
 use super::signal::{
     FreshnessOutcome, build_signal_event, create_traversal_answer, create_traversal_offer,
     estimate_clock_skew, validate_offer_freshness, validate_traversal_answer_for_offer,
@@ -10,6 +10,7 @@ use super::traversal::{
     PunchStrategy, build_punch_packet, parse_punch_packet, plan_punch_targets,
     planned_remote_endpoints, session_hash,
 };
+use super::traversal_machine::suppress_responder_for_own_initiator;
 use super::{
     ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, OverlayAdvert, OverlayEndpointAdvert,
     OverlayTransportKind, PunchHint, PunchPacketKind, TraversalAddress,

src/nostr/traversal.rs

@@ -177,17 +177,15 @@ pub(super) async fn run_punch_attempt(
         let Ok(Ok((len, remote))) = recv else {
             break Err(BootstrapError::PunchTimeout(session_id.to_string()));
         };
-        let Ok(packet) = parse_punch_packet(&buf[..len]) else {
-            continue;
-        };
-        if packet.session_hash != expected_hash {
-            continue;
+        match classify_punch_packet(&buf[..len], expected_hash) {
+            PunchAction::Ignore => continue,
+            PunchAction::Ack { sequence } => {
+                let ack = build_punch_packet(PunchPacketKind::Ack, sequence, session_id);
+                let _ = udp.send_to(&ack, remote).await;
+                break Ok(remote);
+            }
+            PunchAction::Matched => break Ok(remote),
         }
-        if packet.kind == PunchPacketKind::Probe {
-            let ack = build_punch_packet(PunchPacketKind::Ack, packet.sequence, session_id);
-            let _ = udp.send_to(&ack, remote).await;
-        }
-        break Ok(remote);
     };
     send_handle.abort();
     result
@@ -274,3 +272,82 @@ pub(super) fn parse_punch_packet(bytes: &[u8]) -> Result<PunchPacket, BootstrapE
         session_hash: hash,
     })
 }
+
+/// Classification of a received UDP datagram on the punch socket. Returned
+/// by [`classify_punch_packet`]; the timing loop performs the actual ack
+/// send / break described by the variant.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum PunchAction {
+    /// Not a valid punch packet for this session — keep listening.
+    Ignore,
+    /// A matching probe: the driver builds and sends an ack for `sequence`,
+    /// then treats the peer as reached.
+    Ack { sequence: u32 },
+    /// A matching non-probe (ack) packet: the peer is reached, no ack to send.
+    Matched,
+}
+
+/// Pure classification of a received datagram against the expected session
+/// hash. No I/O: the caller sends any ack and decides control flow.
+pub(super) fn classify_punch_packet(bytes: &[u8], expected_hash: [u8; 16]) -> PunchAction {
+    let Ok(packet) = parse_punch_packet(bytes) else {
+        return PunchAction::Ignore;
+    };
+    if packet.session_hash != expected_hash {
+        return PunchAction::Ignore;
+    }
+    if packet.kind == PunchPacketKind::Probe {
+        PunchAction::Ack {
+            sequence: packet.sequence,
+        }
+    } else {
+        PunchAction::Matched
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const SESSION: &str = "session-classify-vectors";
+
+    #[test]
+    fn classify_ignores_unparseable_bytes() {
+        // P1: too short to parse.
+        assert_eq!(
+            classify_punch_packet(&[0u8; 4], session_hash(SESSION)),
+            PunchAction::Ignore
+        );
+    }
+
+    #[test]
+    fn classify_ignores_mismatched_session_hash() {
+        // P2: parseable, but hash is for a different session.
+        let packet = build_punch_packet(PunchPacketKind::Probe, 7, SESSION);
+        let other_hash = session_hash("some-other-session");
+        assert_eq!(
+            classify_punch_packet(&packet, other_hash),
+            PunchAction::Ignore
+        );
+    }
+
+    #[test]
+    fn classify_probe_matching_hash_acks_with_sequence() {
+        // P3: matching probe -> Ack carrying the packet's sequence.
+        let packet = build_punch_packet(PunchPacketKind::Probe, 42, SESSION);
+        assert_eq!(
+            classify_punch_packet(&packet, session_hash(SESSION)),
+            PunchAction::Ack { sequence: 42 }
+        );
+    }
+
+    #[test]
+    fn classify_ack_matching_hash_is_matched() {
+        // P4: matching non-probe (ack) -> Matched.
+        let packet = build_punch_packet(PunchPacketKind::Ack, 3, SESSION);
+        assert_eq!(
+            classify_punch_packet(&packet, session_hash(SESSION)),
+            PunchAction::Matched
+        );
+    }
+}

src/nostr/traversal_machine.rs

@@ -0,0 +1,333 @@
+//! Synchronous decision core for the Nostr NAT-traversal control state.
+//!
+//! `TraversalMachine` owns the engine-scoped, cross-session state that
+//! previously lived directly on `NostrRendezvous` — the set of in-flight
+//! outbound initiators, and the replay/seen-sessions cache — and hosts the
+//! *decisions* over that state: initiator dedup, the dual-`auto_connect`
+//! responder election, and replay rejection.
+//!
+//! Following the sans-IO shape used by `advert` / `failure_state`, every
+//! method here is synchronous, performs no network I/O and no `.await`,
+//! holds its state behind `std::sync::Mutex`, and takes the current time as
+//! an explicit `now_ms: u64` input rather than reading a clock. The async
+//! driver on `NostrRendezvous` reads the clock at the call site, derives the
+//! `NodeAddr`s from npubs, and performs the actual socket/STUN/relay I/O.
+//!
+//! The inherently-async concurrency primitives stay driver-side: the
+//! `pending_answers` oneshot routing and the `offer_slots` semaphore
+//! admission are not modeled here, and neither is the punch send-cadence.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Mutex;
+
+use crate::NodeAddr;
+
+/// Decide whether an incoming-offer responder session should be suppressed
+/// in favour of our own already-running outbound initiator session.
+///
+/// Two peers that each have the other as `auto_connect` simultaneously run an
+/// initiator traversal *and* a responder traversal for the same peer, binding a
+/// separate UDP socket per session. Each node then emits two
+/// `BootstrapEvent::Established` events and `adopt_established_traversal` keeps
+/// only the first on a non-deterministic race; when the two nodes' independent
+/// races resolve to mismatched sessions, each side's Noise msg1 lands on a peer
+/// port the peer already stopped draining and both handshakes stall.
+///
+/// To collapse the four-socket dance to a single, guaranteed-matching socket
+/// pair, both nodes deterministically keep the session **initiated by the
+/// smaller `NodeAddr`** — reusing the project's existing NodeAddr tie-breaker
+/// convention (`cross_connection_winner`, the rekey dual-init resolution, and
+/// the dual-cross-init adopt path in `lifecycle.rs`).
+///
+/// This is evaluated on the responder path, where the session being handled is
+/// *peer-initiated*. It returns `true` (suppress this responder session) only
+/// when genuine duplication exists — i.e. we also have an in-flight outbound
+/// initiator for this same peer (`have_active_initiator`) — and our own
+/// initiator session is the preferred one (`our_addr < peer_addr`). When there
+/// is no co-active initiator (the asymmetric / one-sided `auto_connect` case,
+/// where only one session exists at all) it never suppresses, so connectivity
+/// is preserved. The `our_addr == peer_addr` case (self / loopback) and any
+/// caller that cannot derive a peer `NodeAddr` likewise fall through to "do not
+/// suppress".
+pub(super) fn suppress_responder_for_own_initiator(
+    our_addr: &NodeAddr,
+    peer_addr: &NodeAddr,
+    have_active_initiator: bool,
+) -> bool {
+    have_active_initiator && our_addr < peer_addr
+}
+
+/// Result of the dual-init responder election. Returned by
+/// [`TraversalMachine::classify_incoming_offer`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum OfferDisposition {
+    /// Answer this offer normally.
+    Proceed,
+    /// Decline this responder session; our own outbound initiator wins.
+    Suppress,
+}
+
+/// Result of the replay / seen-sessions check. Returned by
+/// [`TraversalMachine::note_session_seen`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum SeenDecision {
+    /// First time we've seen this session id within the replay window. If a
+    /// cap eviction occurred, `evicted` carries `(evicted, retained)` for the
+    /// driver's overflow debug line.
+    Fresh { evicted: Option<(usize, usize)> },
+    /// The session id is already present within the replay window.
+    Replay,
+}
+
+pub(super) struct TraversalMachine {
+    /// Replay window length in ms (`replay_window_secs * 1000`).
+    replay_window_ms: u64,
+    /// Size cap for the seen-sessions cache.
+    seen_max_entries: usize,
+    /// npubs of peers we currently have an in-flight outbound initiator for.
+    active_initiators: Mutex<HashSet<String>>,
+    /// Replay cache: session id -> expiry (ms).
+    seen_sessions: Mutex<HashMap<String, u64>>,
+}
+
+impl TraversalMachine {
+    pub(super) fn new(replay_window_ms: u64, seen_max_entries: usize) -> Self {
+        Self {
+            replay_window_ms,
+            seen_max_entries,
+            active_initiators: Mutex::new(HashSet::new()),
+            seen_sessions: Mutex::new(HashMap::new()),
+        }
+    }
+
+    // --- initiator lifecycle -------------------------------------------
+
+    /// Register an in-flight outbound initiator for `npub`. Returns `false`
+    /// when one is already active (dedup — the driver should not start a
+    /// second).
+    pub(super) fn begin_initiator(&self, npub: &str) -> bool {
+        self.lock_initiators().insert(npub.to_string())
+    }
+
+    /// Clear the in-flight initiator for `npub` on task completion.
+    pub(super) fn end_initiator(&self, npub: &str) {
+        self.lock_initiators().remove(npub);
+    }
+
+    // --- responder dual-init election ----------------------------------
+
+    /// Decide whether an incoming offer from `sender_npub` should be answered
+    /// or suppressed in favour of our own outbound initiator. The driver
+    /// derives both `NodeAddr`s (keeping today's derivation-failure "answer
+    /// anyway" fallthrough) and passes them in; the machine reads its own
+    /// active-initiator membership under its lock.
+    pub(super) fn classify_incoming_offer(
+        &self,
+        sender_npub: &str,
+        our_addr: &NodeAddr,
+        peer_addr: &NodeAddr,
+    ) -> OfferDisposition {
+        let have_active = self.lock_initiators().contains(sender_npub);
+        if suppress_responder_for_own_initiator(our_addr, peer_addr, have_active) {
+            OfferDisposition::Suppress
+        } else {
+            OfferDisposition::Proceed
+        }
+    }
+
+    // --- replay / seen-sessions ----------------------------------------
+
+    /// Record that `session_id` was seen at `now_ms`. Prunes expired entries,
+    /// rejects a replay, inserts the fresh id, then applies the size cap.
+    ///
+    /// Returns [`SeenDecision::Replay`] in place of the old
+    /// `Err(BootstrapError::Replay)`, or [`SeenDecision::Fresh`] carrying the
+    /// `(evicted, retained)` pair when a cap eviction occurred.
+    pub(super) fn note_session_seen(&self, session_id: &str, now_ms: u64) -> SeenDecision {
+        let expiry = now_ms + self.replay_window_ms;
+        let mut seen = self.lock_seen();
+        seen.retain(|_, expires_at| *expires_at > now_ms);
+        if seen.contains_key(session_id) {
+            return SeenDecision::Replay;
+        }
+        seen.insert(session_id.to_string(), expiry);
+        if seen.len() > self.seen_max_entries {
+            let mut oldest = seen
+                .iter()
+                .map(|(session, expires_at)| (session.clone(), *expires_at))
+                .collect::<Vec<_>>();
+            oldest.sort_by_key(|(_, expires_at)| *expires_at);
+            let overflow = seen.len().saturating_sub(self.seen_max_entries);
+            for (session, _) in oldest.into_iter().take(overflow) {
+                seen.remove(&session);
+            }
+            return SeenDecision::Fresh {
+                evicted: Some((overflow, seen.len())),
+            };
+        }
+        SeenDecision::Fresh { evicted: None }
+    }
+
+    // --- lock helpers ---------------------------------------------------
+
+    fn lock_initiators(&self) -> std::sync::MutexGuard<'_, HashSet<String>> {
+        self.active_initiators
+            .lock()
+            .expect("traversal-machine active-initiators mutex poisoned")
+    }
+
+    fn lock_seen(&self) -> std::sync::MutexGuard<'_, HashMap<String, u64>> {
+        self.seen_sessions
+            .lock()
+            .expect("traversal-machine seen-sessions mutex poisoned")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn node_addr(first_byte: u8) -> NodeAddr {
+        let mut bytes = [0u8; 16];
+        bytes[0] = first_byte;
+        NodeAddr::from_bytes(bytes)
+    }
+
+    fn machine() -> TraversalMachine {
+        TraversalMachine::new(1_000_000, 3)
+    }
+
+    // --- election ordering (classify_incoming_offer) -------------------
+
+    #[test]
+    fn election_vectors() {
+        let smaller = node_addr(0x01);
+        let larger = node_addr(0x02);
+        let peer = "npub1peer";
+
+        // V1: co-active initiator, our addr smaller -> Suppress.
+        let m = machine();
+        assert!(m.begin_initiator(peer));
+        assert_eq!(
+            m.classify_incoming_offer(peer, &smaller, &larger),
+            OfferDisposition::Suppress
+        );
+
+        // V2: co-active initiator, our addr larger -> Proceed.
+        let m = machine();
+        assert!(m.begin_initiator(peer));
+        assert_eq!(
+            m.classify_incoming_offer(peer, &larger, &smaller),
+            OfferDisposition::Proceed
+        );
+
+        // V3: no co-active initiator, our addr smaller -> Proceed
+        // (asymmetric one-sided auto_connect).
+        let m = machine();
+        assert_eq!(
+            m.classify_incoming_offer(peer, &smaller, &larger),
+            OfferDisposition::Proceed
+        );
+
+        // V4: co-active initiator, equal addresses -> Proceed (self/loopback).
+        let m = machine();
+        assert!(m.begin_initiator(peer));
+        assert_eq!(
+            m.classify_incoming_offer(peer, &smaller, &smaller),
+            OfferDisposition::Proceed
+        );
+
+        // V5: classify BEFORE begin_initiator -> Proceed (offer-first race).
+        let m = machine();
+        assert_eq!(
+            m.classify_incoming_offer(peer, &smaller, &larger),
+            OfferDisposition::Proceed
+        );
+
+        // V6: begin then end then classify -> Proceed (initiator finished).
+        let m = machine();
+        assert!(m.begin_initiator(peer));
+        m.end_initiator(peer);
+        assert_eq!(
+            m.classify_incoming_offer(peer, &smaller, &larger),
+            OfferDisposition::Proceed
+        );
+    }
+
+    // --- initiator dedup ------------------------------------------------
+
+    #[test]
+    fn initiator_dedup() {
+        let m = machine();
+        assert!(m.begin_initiator("npub1p"), "first is a fresh initiator");
+        assert!(
+            !m.begin_initiator("npub1p"),
+            "second for same npub is a dup"
+        );
+        m.end_initiator("npub1p");
+        assert!(
+            m.begin_initiator("npub1p"),
+            "fresh again after end_initiator"
+        );
+    }
+
+    // --- replay (note_session_seen) ------------------------------------
+
+    #[test]
+    fn replay_first_then_repeat() {
+        // R1: first id Fresh; same id within window Replay.
+        let m = machine();
+        assert_eq!(
+            m.note_session_seen("s1", 1000),
+            SeenDecision::Fresh { evicted: None }
+        );
+        assert_eq!(m.note_session_seen("s1", 1500), SeenDecision::Replay);
+    }
+
+    #[test]
+    fn replay_prunes_expired() {
+        // R2: an entry past its expiry is pruned, so re-seeing it is Fresh.
+        let m = machine(); // replay_window_ms = 1_000_000
+        assert_eq!(
+            m.note_session_seen("s1", 1000),
+            SeenDecision::Fresh { evicted: None }
+        );
+        // now well past s1's expiry (1000 + 1_000_000): s1 pruned by retain,
+        // so s1 is Fresh again rather than Replay.
+        assert_eq!(
+            m.note_session_seen("s1", 5_000_000),
+            SeenDecision::Fresh { evicted: None }
+        );
+    }
+
+    #[test]
+    fn replay_cap_evicts_oldest_by_expiry() {
+        // R3: cap overflow evicts oldest-by-expiry, returns (evicted, retained).
+        let m = machine(); // cap = 3, window huge so nothing expires here
+        assert_eq!(
+            m.note_session_seen("s1", 1),
+            SeenDecision::Fresh { evicted: None }
+        );
+        assert_eq!(
+            m.note_session_seen("s2", 2),
+            SeenDecision::Fresh { evicted: None }
+        );
+        assert_eq!(
+            m.note_session_seen("s3", 3),
+            SeenDecision::Fresh { evicted: None }
+        );
+        assert_eq!(
+            m.note_session_seen("s4", 4),
+            SeenDecision::Fresh {
+                evicted: Some((1, 3))
+            }
+        );
+        // Oldest expiry (s1) was evicted, so re-seeing it is Fresh.
+        assert_eq!(
+            m.note_session_seen("s1", 5),
+            SeenDecision::Fresh {
+                evicted: Some((1, 3))
+            }
+        );
+    }
+}