From e064c96df35d95e1e0b617c3204d141353dba973 Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 03:37:31 +0000
Subject: [PATCH 1/6] peer: route peer-loss reports by kind

Add a LostKind discriminator to the ReportLost action so the executor can
route an un-promoted handshake failure to the connected-guarded reconnect
reflex (note_handshake_timeout) and an established peer's link-death to the
unconditional one (note_link_dead), instead of collapsing every loss to
note_link_dead.

The two loss producers dispatched today, the liveness reap and the
inbound-restart-then-promote arm, both keep the link-dead routing, so
behavior is unchanged. The handshake-timeout and dial-failure producers are
tagged accordingly but stay dormant until their events are dispatched.
---
 src/node/dataplane/peer_actions.rs | 28 +++++++------
 src/peer/machine.rs                | 65 +++++++++++++++++++++++++-----
 2 files changed, 72 insertions(+), 21 deletions(-)

diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs
index 8ee9496..b19d18a 100644
--- a/src/node/dataplane/peer_actions.rs
+++ b/src/node/dataplane/peer_actions.rs
@@ -19,14 +19,14 @@
 //! realized as those planes are wired).
 //! `RegisterDecryptSession` is a deliberate no-op — see its arm for the note.
 
+use crate::PeerIdentity;
 use crate::node::Node;
 use crate::node::reject::{HandshakeReject, RejectReason};
-use crate::peer::machine::{PeerAction, PeerEvent};
+use crate::peer::machine::{LostKind, PeerAction, PeerEvent};
 use crate::proto::fmp::PromotionResult;
 use crate::proto::fmp::wire::build_msg2;
 use crate::transport::{LinkId, TransportAddr, TransportId};
 use crate::utils::index::SessionIndex;
-use crate::{NodeAddr, PeerIdentity};
 use std::collections::VecDeque;
 use tracing::{debug, trace, warn};
 
@@ -449,18 +449,22 @@ impl Node {
                     // INERT — the legacy tick timers still run, so driving
                     // these would double-schedule.
                 }
-                PeerAction::ReportLost { peer } => {
-                    // The single loss token → the reconciler reflex (`driver.rs:48`).
-                    self.report_peer_lost(peer, ambient.now_ms);
+                PeerAction::ReportLost { peer, kind } => {
+                    // The single loss token, routed to the reconciler reflex the
+                    // `kind` names: an un-promoted handshake attempt takes the
+                    // connected-guarded `note_handshake_timeout` (`driver.rs:28`),
+                    // an established peer's link-death takes the unconditional
+                    // `note_link_dead` (`driver.rs:48`).
+                    match kind {
+                        LostKind::HandshakeTimeout => {
+                            self.note_handshake_timeout(peer, ambient.now_ms);
+                        }
+                        LostKind::LinkDead => {
+                            self.note_link_dead(peer, ambient.now_ms);
+                        }
+                    }
                 }
             }
         }
     }
-
-    /// `ReportLost` → `note_link_dead` (kept as a named seam so the ambient clock
-    /// source is explicit and the reconciler-computed backoff can be threaded later).
-    #[allow(dead_code)]
-    fn report_peer_lost(&mut self, peer: NodeAddr, now_ms: u64) {
-        self.note_link_dead(peer, now_ms);
-    }
 }
diff --git a/src/peer/machine.rs b/src/peer/machine.rs
index d4aeefc..3110eaf 100644
--- a/src/peer/machine.rs
+++ b/src/peer/machine.rs
@@ -257,6 +257,21 @@ pub(crate) enum PeerEvent {
     Tick,
 }
 
+/// Why a peer was reported lost. Selects the reconciler reflex the executor
+/// routes the `ReportLost` token to: an un-promoted handshake attempt that
+/// failed (`HandshakeTimeout`, connected-guarded like the old `schedule_retry`)
+/// versus an established peer whose link died (`LinkDead`, unconditional like
+/// the old `schedule_reconnect`).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum LostKind {
+    /// An outbound handshake attempt timed out or its dial failed before the
+    /// peer promoted — routes to the connected-guarded reflex.
+    HandshakeTimeout,
+    /// An established peer's link went dead or is being replaced — routes to
+    /// the unconditional reconnect reflex.
+    LinkDead,
+}
+
 /// An effect the driver executes on the machine's behalf. Runtime-agnostic
 /// plain data — no tokio handles, time only as `at_ms` fields.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -303,8 +318,9 @@ pub(crate) enum PeerAction {
     /// Cancel a scheduled timer.
     CancelTimer { kind: TimerKind },
     /// Report the peer lost to the reconciler (the single loss token — there is
-    /// deliberately no `ScheduleRetry` machine action).
-    ReportLost { peer: NodeAddr },
+    /// deliberately no `ScheduleRetry` machine action). `kind` selects the
+    /// reflex (handshake-timeout vs link-dead) the executor routes to.
+    ReportLost { peer: NodeAddr, kind: LostKind },
 }
 
 // ============================================================================
@@ -527,7 +543,13 @@ impl PeerMachine {
         }
         let mut actions = Vec::new();
         if let Some(peer) = self.addr() {
-            actions.push(PeerAction::ReportLost { peer });
+            // Dial failure on an un-promoted leg routes like a handshake timeout
+            // (the connected-guarded reflex). Dormant today — no `TransportFailed`
+            // event is dispatched until the connection-oriented cutover (C5).
+            actions.push(PeerAction::ReportLost {
+                peer,
+                kind: LostKind::HandshakeTimeout,
+            });
         }
         self.state = PeerState::Closed {
             backoff_deadline_ms: now + CLOSED_BACKOFF_MS,
@@ -641,7 +663,12 @@ impl PeerMachine {
                 if let Some(idx) = self.our_index.take() {
                     actions.push(PeerAction::UnregisterDecryptSession { index: idx });
                 }
-                actions.push(PeerAction::ReportLost { peer });
+                // An established peer being replaced by a fresh inbound leg — the
+                // unconditional reconnect reflex (a live, cut-over producer).
+                actions.push(PeerAction::ReportLost {
+                    peer,
+                    kind: LostKind::LinkDead,
+                });
                 actions.extend(self.inbound_classify(link, &wire));
                 actions
             }
@@ -1017,7 +1044,12 @@ impl PeerMachine {
             PeerAction::TeardownConnectedUdp,
         ];
         if let Some(peer) = self.addr() {
-            actions.push(PeerAction::ReportLost { peer });
+            // An established peer whose link died — the unconditional reconnect
+            // reflex (the live liveness-reap producer).
+            actions.push(PeerAction::ReportLost {
+                peer,
+                kind: LostKind::LinkDead,
+            });
         }
         self.state = PeerState::Closed {
             backoff_deadline_ms: now + CLOSED_BACKOFF_MS,
@@ -1088,7 +1120,13 @@ impl PeerMachine {
         for act in Fmp::new().poll_timeouts(vec![snap]) {
             match act {
                 ConnAction::ScheduleRetry { peer } => {
-                    lost.push(PeerAction::ReportLost { peer });
+                    // Handshake timeout on an un-promoted leg — the connected-
+                    // guarded reflex. Dormant today (no `Timeout` event is
+                    // dispatched until the timeout fold in C5).
+                    lost.push(PeerAction::ReportLost {
+                        peer,
+                        kind: LostKind::HandshakeTimeout,
+                    });
                 }
                 ConnAction::Teardown { .. } => {
                     if let Some(idx) = self.conn.our_index() {
@@ -1338,7 +1376,10 @@ mod tests {
             PeerAction::CancelTimer {
                 kind: TimerKind::Liveness,
             },
-            PeerAction::ReportLost { peer },
+            PeerAction::ReportLost {
+                peer,
+                kind: LostKind::LinkDead,
+            },
         ];
         for a in &sample {
             // Exhaustiveness guard: no `_` wildcard, so adding a variant without
@@ -1587,7 +1628,10 @@ mod tests {
                 PeerAction::UnregisterDecryptSession {
                     index: SessionIndex::new(0xDEAD)
                 },
-                PeerAction::ReportLost { peer: peer_addr },
+                PeerAction::ReportLost {
+                    peer: peer_addr,
+                    kind: LostKind::LinkDead,
+                },
             ]
         );
         assert!(matches!(
@@ -1988,7 +2032,10 @@ mod tests {
             vec![
                 PeerAction::InvalidateSendState,
                 PeerAction::TeardownConnectedUdp,
-                PeerAction::ReportLost { peer: addr },
+                PeerAction::ReportLost {
+                    peer: addr,
+                    kind: LostKind::LinkDead,
+                },
             ]
         );
         assert!(matches!(m.state(), PeerState::Closed { .. }));

From 3b99a416ad87223836b8c98794b12451c7752a2b Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 04:06:44 +0000
Subject: [PATCH 2/6] peer: persist the outbound control machine at dial

Create and persist the per-peer control machine when an outbound handshake is
dialed, keyed by its link, instead of building a transient at msg2. The msg2
completion path now looks up that persisted machine to drive the promote,
falling back to a transient only if none is present (e.g. a direct-seeded test).

The machine parks in the Discovered state until promotion and is inert to the
liveness reap and rekey cadence while unpromoted, since it is absent from the
peers map. It is removed on every path that ends the outbound leg without
promoting -- the stale-connection reaper, the msg2 authorization-failure arm,
and the cross-connection resolution block -- mirroring the connection's own
lifetime so no dangling machine survives.

Its session index is deliberately left unset on the machine (the shell owns the
index on its connection), so a later inbound restart does not emit a spurious
decrypt-session unregister. A regression test covers that invariant.
---
 src/node/dataplane/peer_actions.rs |  7 +--
 src/node/handlers/handshake.rs     | 84 ++++++++++++++++-----------
 src/node/handlers/timeout.rs       |  5 ++
 src/node/lifecycle/mod.rs          | 13 +++++
 src/peer/machine.rs                | 91 ++++++++++++++++++++++++++++++
 5 files changed, 164 insertions(+), 36 deletions(-)

diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs
index b19d18a..fcfa376 100644
--- a/src/node/dataplane/peer_actions.rs
+++ b/src/node/dataplane/peer_actions.rs
@@ -292,10 +292,9 @@ impl Node {
                                 // pre-refactor outbound warn! ("Failed to promote
                                 // connection").
                                 //
-                                // The transient outbound machine was inserted BEFORE
-                                // execute; it is additive state that
-                                // did not exist pre-refactor, so removing the just-
-                                // inserted machine on failure is neutral vs old and
+                                // The outbound machine was persisted at dial; it is
+                                // additive state that did not exist pre-refactor, so
+                                // removing it on promote failure is neutral vs old and
                                 // prevents a leak.
                                 warn!(
                                     target: "fips::node::handlers::handshake",
diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs
index 45e890e..345dfc9 100644
--- a/src/node/handlers/handshake.rs
+++ b/src/node/handlers/handshake.rs
@@ -973,6 +973,8 @@ impl Node {
                 }
             }
             self.connections.remove(&link_id);
+            // Drop the machine persisted at dial — this leg never promotes.
+            self.peer_machines.remove(&link_id);
             self.remove_link(&link_id);
             if let Some(idx) = our_index {
                 let _ = self.index_allocator.free(idx);
@@ -1007,6 +1009,12 @@ impl Node {
         let out_snap = self.outbound_snapshot(&peer_node_addr);
         let out_decision = self.fmp.establish_outbound(&out_snap);
         if out_decision != OutboundDecision::Promote {
+            // The dial-persisted outbound machine is not consumed by the inline
+            // Swap/Keep resolution below (which mutates the existing promoted peer
+            // 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);
             // Extract the outbound connection
             let mut conn = match self.connections.remove(&link_id) {
                 Some(c) => c,
@@ -1132,43 +1140,55 @@ impl Node {
         // loser-link surgery is wired later). Direct analog of the inbound
         // net-new arm — no ordering constraint, lowest risk.
         //
-        // Build a TRANSIENT outbound machine, step `Msg2 →
-        // [PromoteToActive]`, execute it (→ `promote_connection` →
-        // `PromotionResolved{Promoted}` → inert `RegisterDecryptSession`), and
-        // insert into `peer_machines` only on the Promoted (Established) tail. The
-        // outbound `our_index` was allocated at DIAL (unchanged), the outbound
-        // promote sends nothing on the wire, and `promote_connection` frees
-        // nothing new — so the index sequence, `peers`/`peers_by_index`/
-        // `addr_to_link` state, and metrics are byte-identical to the pre-refactor
-        // Promoted arm. `pending_outbound` lifecycle stays shell-side (removed on
-        // the Established tail, exactly where the pre-refactor Ok arm removed it);
-        // the machine never touches it.
-        let mut machine = PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms);
-
-        // Step `Msg2 → [PromoteToActive]`. The machine re-runs the pure
-        // `establish_outbound` on the snapshot (a harmless second pure call);
-        // `has_existing_peer == false` reproduces the `Promote` decision.
-        let promote_actions = machine.step(
-            PeerEvent::Msg2 {
-                their_index: header.sender_idx,
-                out: out_snap,
-            },
-            packet.timestamp_ms,
-            &mut self.index_allocator,
-        );
+        // Look up the outbound machine persisted at DIAL (`start_handshake`),
+        // parked in `Discovered`, and step `Msg2 → [PromoteToActive]` in place.
+        // `on_msg2` is state-independent — it decides from the outbound snapshot,
+        // not the machine's state — and reads the machine's unset `conn.our_index`
+        // as `None`, so stepping the persisted (vs the former transient) machine
+        // is byte-identical: same `Promote` decision (`has_existing_peer == false`),
+        // same `our_index == None`. The machine is already in `peer_machines`
+        // (inserted at dial), so the executor's `PromoteToActive` arm can feed
+        // `PromotionResolved` back via the same lookup. A defensive transient
+        // reproduces the pre-persistence path if the machine is somehow absent — a
+        // state-machine inconsistency, since every dialed leg persists one. The
+        // outbound `our_index` was allocated at DIAL (unchanged), the promote
+        // sends nothing on the wire, and `promote_connection` frees nothing new,
+        // so index sequence, `peers`/`peers_by_index`/`addr_to_link` state, and
+        // metrics are byte-identical. `pending_outbound` lifecycle stays shell-side
+        // (removed on the Established tail); the machine never touches it.
+        let promote_actions = match self.peer_machines.get_mut(&link_id) {
+            Some(machine) => machine.step(
+                PeerEvent::Msg2 {
+                    their_index: header.sender_idx,
+                    out: out_snap,
+                },
+                packet.timestamp_ms,
+                &mut self.index_allocator,
+            ),
+            None => {
+                // No machine persisted at dial (e.g. a test that seeds
+                // `connections`/`pending_outbound` directly, or any path that
+                // reaches msg2 without `start_handshake`): reproduce the
+                // pre-persistence transient exactly.
+                let mut machine =
+                    PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms);
+                let actions = machine.step(
+                    PeerEvent::Msg2 {
+                        their_index: header.sender_idx,
+                        out: out_snap,
+                    },
+                    packet.timestamp_ms,
+                    &mut self.index_allocator,
+                );
+                self.peer_machines.insert(link_id, machine);
+                actions
+            }
+        };
         debug_assert_eq!(
             promote_actions,
             vec![PeerAction::PromoteToActive { link: link_id }]
         );
 
-        // Register the machine (Promote tail only — the Swap/Keep/rekey arms above
-        // all returned without inserting). Inserted BEFORE execute so the
-        // executor's `PromoteToActive` arm can feed `PromotionResolved` back into
-        // it via the `peer_machines` lookup. The outbound `link_id` was allocated
-        // at dial and the dial path never inserts a machine, so this
-        // cannot collide with an existing entry.
-        self.peer_machines.insert(link_id, machine);
-
         // Execute `[PromoteToActive]`. The executor calls `promote_connection`,
         // feeds `PromotionResolved{Promoted}` back, registers the decrypt-worker
         // session (the register was relocated into the executor's
diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs
index af88741..c1af290 100644
--- a/src/node/handlers/timeout.rs
+++ b/src/node/handlers/timeout.rs
@@ -115,6 +115,11 @@ impl Node {
             Some(c) => c,
             None => return,
         };
+        // A dial-persisted outbound control machine shares the connection's
+        // `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);
         let transport_id = conn.transport_id();
 
         // Free session index and pending_outbound if allocated
diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs
index 5536bb3..6c84e45 100644
--- a/src/node/lifecycle/mod.rs
+++ b/src/node/lifecycle/mod.rs
@@ -15,6 +15,7 @@ use crate::node::acl::PeerAclContext;
 use crate::nostr::{BootstrapEvent, NostrRendezvous};
 use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal};
 use crate::peer::PeerConnection;
+use crate::peer::machine::PeerMachine;
 use crate::proto::fmp::wire::build_msg1;
 use crate::proto::fmp::{Disconnect, DisconnectReason};
 use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
@@ -582,6 +583,18 @@ impl Node {
             .insert((transport_id, our_index.as_u32()), link_id);
         self.connections.insert(link_id, connection);
 
+        // Persist the outbound control machine at dial, keyed by the same
+        // `link_id` as the connection. It parks in `Discovered` (inert to reap
+        // and rekey — it is absent from `peers`, never established) until
+        // `handle_msg2` looks it up to drive the promote. Its `our_index` is
+        // deliberately left unset so a later inbound restart does not emit a
+        // spurious `UnregisterDecryptSession`. It is removed wherever the
+        // connection is torn down without promoting (the stale reaper and the
+        // msg2 ACL-fail / cross-connection arms), mirroring the connection's
+        // own lifetime.
+        let machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms);
+        self.peer_machines.insert(link_id, machine);
+
         // Send the wire format handshake message
         if let Some(transport) = self.transports.get(&transport_id) {
             match transport.send(&remote_addr, &wire_msg1).await {
diff --git a/src/peer/machine.rs b/src/peer/machine.rs
index 3110eaf..e4f0ba8 100644
--- a/src/peer/machine.rs
+++ b/src/peer/machine.rs
@@ -2002,6 +2002,97 @@ mod tests {
         );
     }
 
+    // ---- Test 7b: dial-persisted outbound promote leaves our_index unset ---
+    // An outbound machine persisted at DIAL (`new_outbound`, `Discovered`, with
+    // `conn.our_index` UNSET — the shell owns the index on its own
+    // `PeerConnection`, never on the machine) must, on promote via msg2, end with
+    // `our_index == None`, exactly as the pre-persistence transient did. The
+    // guard: a subsequent inbound restart then emits NO
+    // `UnregisterDecryptSession` (contrast `restart_override`, whose machine has
+    // `our_index == Some`). A leaked `Some(dial_index)` here would wrongly
+    // unregister — on index reuse, ANOTHER peer's — worker session; keeping the
+    // field `None` is the Model-B dial-persistence neutrality property.
+    #[test]
+    fn dial_persisted_outbound_promote_no_restart_unregister() {
+        let mut alloc = IndexAllocator::new();
+        let peer = peer_identity();
+        let peer_addr = *peer.node_addr();
+        let our = *peer_identity().node_addr();
+
+        // Persisted at dial: Discovered, conn.our_index deliberately NOT set.
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0);
+        assert_eq!(m.our_index(), None);
+
+        // Promote via msg2 from Discovered (the production path — the former
+        // transient was likewise stepped from `new_outbound` without a state set).
+        let out = OutboundSnapshot {
+            has_existing_peer: false,
+            our_outbound_wins: false,
+        };
+        let promote = m.step(
+            PeerEvent::Msg2 {
+                their_index: SessionIndex::new(0x77),
+                out,
+            },
+            300,
+            &mut alloc,
+        );
+        assert_eq!(
+            promote,
+            vec![PeerAction::PromoteToActive {
+                link: LinkId::new(1)
+            }]
+        );
+        assert_eq!(
+            m.our_index(),
+            None,
+            "outbound promote must leave our_index unset"
+        );
+
+        // Drive promotion to Established (from Discovered, as in production).
+        let _ = m.step(
+            PeerEvent::PromotionResolved {
+                result: PromotionResult::Promoted(peer_addr),
+            },
+            300,
+            &mut alloc,
+        );
+        assert_eq!(m.state(), PeerState::Established { addr: peer_addr });
+        assert_eq!(m.our_index(), None);
+
+        // A subsequent inbound restart (peer restart, new epoch) must NOT emit
+        // UnregisterDecryptSession, because our_index is None.
+        let mut est = est_new_peer(our);
+        est.has_existing_peer = true;
+        est.existing_peer_epoch = Some([1u8; 8]);
+        let wire = wire_outcome(peer, Some([2u8; 8]), 0x88);
+        let restart = m.step(
+            PeerEvent::InboundMsg1 {
+                link: LinkId::new(1),
+                wire,
+                est,
+            },
+            1_000,
+            &mut alloc,
+        );
+        assert!(
+            !restart
+                .iter()
+                .any(|a| matches!(a, PeerAction::UnregisterDecryptSession { .. })),
+            "no UnregisterDecryptSession when the promoted outbound machine's our_index is None"
+        );
+        assert!(
+            restart.iter().any(|a| matches!(
+                a,
+                PeerAction::ReportLost {
+                    kind: LostKind::LinkDead,
+                    ..
+                }
+            )),
+            "restart still reports the loss via the link-dead reconnect reflex"
+        );
+    }
+
     // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost --------------
     #[test]
     fn liveness_to_link_dead() {

From 1f765cfd8f1bc9491efa9b68935432496ea640be Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 06:15:21 +0000
Subject: [PATCH 3/6] peer: drive the connectionless outbound msg1 send through
 the machine

Split the outbound handshake setup out of start_handshake into
prepare_outbound_msg1 (allocate the index, run the Noise leaf, frame and arm
msg1 -- the fallible steps, returning an error the caller propagates) and
send_stored_msg1 (transmit the armed wire). A connectionless dial now runs
prepare in the shell, then drives the control machine, whose SendHandshake
action sends the wire via the executor. Connection-oriented dials keep calling
start_handshake, now prepare followed by the send inline.

The dial event gains a connection_oriented flag so the machine's on_dial sends
msg1 immediately for connectionless transports (no connect step) instead of
opening a transport first. Behavior is unchanged: the same index, Noise leaf,
wire bytes, maps, and send-error handling as before, with index-allocation and
Noise failures still propagated synchronously before the send. A regression
test covers that a promote from the post-dial handshaking state is identical to
the former discovered-state promote.
---
 src/node/dataplane/peer_actions.rs |  44 +++++++----
 src/node/handlers/handshake.rs     |  10 ++-
 src/node/lifecycle/mod.rs          | 115 ++++++++++++++++++++++++-----
 src/peer/machine.rs                |  99 ++++++++++++++++++++++---
 4 files changed, 218 insertions(+), 50 deletions(-)

diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs
index fcfa376..74b99e8 100644
--- a/src/node/dataplane/peer_actions.rs
+++ b/src/node/dataplane/peer_actions.rs
@@ -6,18 +6,19 @@
 //! stands for (`build_msg2` + `transport.send`, `promote_connection`,
 //! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …).
 //!
-//! ## Shadow-only skeleton
+//! ## Progressive cutover
 //!
-//! The machine home (`Node.peer_machines`), the
-//! executor, and the disjoint-borrow advance helper. It is **unwired** — the live
-//! `handle_msg1`/`handle_msg2` path does not drive it yet, so every method here is
-//! `#[allow(dead_code)]`. The inbound cutover (`handle_msg1` → `step(InboundMsg1)`)
-//! and the outbound cutover (`handle_msg2` / dial) are wired later.
+//! The executor is wired incrementally. Live today: the inbound establish
+//! (`handle_msg1` → `step(InboundMsg1)`), the outbound msg2 promote
+//! (`handle_msg2` looks up the dial-persisted machine), and the connectionless
+//! outbound msg1 send (`SendHandshake` with `their_index == None` →
+//! `send_stored_msg1`, driven from `initiate_connection`).
 //!
-//! Arms not yet exercised are inert stubs (outbound dial, rekey/crypto installs,
-//! link-control frames, timers, and the connected-UDP plane are inert stubs
-//! realized as those planes are wired).
-//! `RegisterDecryptSession` is a deliberate no-op — see its arm for the note.
+//! Not yet driven, so their arms stay inert stubs: `OpenTransport` (the
+//! connection-oriented dial), 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.
 
 use crate::PeerIdentity;
 use crate::node::Node;
@@ -114,11 +115,16 @@ impl Node {
                     // yet; inert in the shadow-only skeleton.
                 }
                 PeerAction::SendHandshake { bytes } => {
-                    // The machine payload is the UNFRAMED Noise msg2 payload;
-                    // frame it with our/their index (mirrors `handshake.rs:472`'s
-                    // `build_msg2(our_index, their_index, &payload)`) before the
-                    // wire send. A fresh-outbound msg1 (empty payload → build msg1
-                    // from indices) is framed differently and is wired later.
+                    // Two outbound directions share this action, discriminated by
+                    // `their_index`:
+                    //   msg2 (`their_index == Some`): the machine payload is the
+                    //   UNFRAMED Noise msg2; frame it with our/their index
+                    //   (`build_msg2`) and send.
+                    //   msg1 (`their_index == None`): a fresh outbound handshake;
+                    //   the machine's empty payload is ignored — the shell already
+                    //   allocated the index, ran the Noise leaf, and armed the
+                    //   wire at dial (`prepare_outbound_msg1`); this just sends the
+                    //   stored wire (see `send_stored_msg1`).
                     if let (Some(sender_idx), Some(receiver_idx)) =
                         (ambient.our_index, ambient.their_index)
                     {
@@ -152,6 +158,14 @@ impl Node {
                                 .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
                             return;
                         }
+                    } else {
+                        // msg1: the shell already allocated the index, ran the
+                        // Noise leaf, and armed the wire on the connection at dial
+                        // (`prepare_outbound_msg1`); send the stored wire. The
+                        // machine's empty payload is ignored.
+                        let _ = bytes;
+                        self.send_stored_msg1(link, ambient.transport_id, &ambient.remote_addr)
+                            .await;
                     }
                 }
                 PeerAction::SendRekey { .. } => {
diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs
index 345dfc9..3f78903 100644
--- a/src/node/handlers/handshake.rs
+++ b/src/node/handlers/handshake.rs
@@ -1140,10 +1140,12 @@ impl Node {
         // loser-link surgery is wired later). Direct analog of the inbound
         // net-new arm — no ordering constraint, lowest risk.
         //
-        // Look up the outbound machine persisted at DIAL (`start_handshake`),
-        // parked in `Discovered`, and step `Msg2 → [PromoteToActive]` in place.
-        // `on_msg2` is state-independent — it decides from the outbound snapshot,
-        // not the machine's state — and reads the machine's unset `conn.our_index`
+        // Look up the outbound machine persisted at DIAL, and step `Msg2 →
+        // [PromoteToActive]` in place. It is in `Discovered` (connection-oriented
+        // dial, not yet cut over) or `Handshaking{SentMsg1}` (connectionless dial,
+        // which drives the machine to send msg1) — either way `on_msg2` is
+        // state-independent: it decides from the outbound snapshot, not the
+        // machine's state, and reads the machine's unset `conn.our_index`
         // as `None`, so stepping the persisted (vs the former transient) machine
         // is byte-identical: same `Promote` decision (`has_existing_peer == false`),
         // same `our_index == None`. The machine is already in `peer_machines`
diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs
index 6c84e45..2f32a14 100644
--- a/src/node/lifecycle/mod.rs
+++ b/src/node/lifecycle/mod.rs
@@ -12,10 +12,11 @@ use super::peering::retry::MAX_RETRY_CONNECTIONS_PER_TICK;
 
 use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
 use crate::node::acl::PeerAclContext;
+use crate::node::dataplane::PeerActionCtx;
 use crate::nostr::{BootstrapEvent, NostrRendezvous};
 use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal};
 use crate::peer::PeerConnection;
-use crate::peer::machine::PeerMachine;
+use crate::peer::machine::{PeerEvent, PeerMachine};
 use crate::proto::fmp::wire::build_msg1;
 use crate::proto::fmp::{Disconnect, DisconnectReason};
 use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
@@ -509,22 +510,71 @@ impl Node {
             }
             Ok(())
         } else {
-            // Connectionless: proceed with immediate handshake
-            self.start_handshake(link_id, transport_id, remote_addr, peer_identity)
-                .await
+            // Connectionless: no connect step. Prepare msg1 in the shell — the
+            // index alloc, Noise leaf, and framing can fail and the error must
+            // propagate to the caller (matching the pre-cutover path) — then drive
+            // the machine to send it. The machine goes straight to
+            // `start_outbound_handshake`; the executor's `SendHandshake` msg1
+            // branch sends the wire `prepare_outbound_msg1` armed on the
+            // connection.
+            self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?;
+            let now = Self::now_ms();
+            let ambient = PeerActionCtx {
+                verified_identity: peer_identity,
+                transport_id,
+                remote_addr: remote_addr.clone(),
+                our_index: None,
+                their_index: None,
+                now_ms: now,
+                is_outbound: true,
+            };
+            self.advance_peer_machine(
+                link_id,
+                PeerEvent::Dial {
+                    transport_id,
+                    remote_addr,
+                    peer_identity,
+                    connection_oriented: false,
+                },
+                now,
+                &ambient,
+            )
+            .await;
+            Ok(())
         }
     }
 
     /// Start the Noise handshake on a link and send msg1.
     ///
-    /// Called immediately for connectionless transports, or after the
-    /// transport connection is established for connection-oriented transports.
+    /// Called after a connection-oriented transport connects. (Connectionless
+    /// dials `prepare_outbound_msg1` + drive the machine to send in
+    /// `initiate_connection`.)
     pub(super) async fn start_handshake(
         &mut self,
         link_id: LinkId,
         transport_id: TransportId,
         remote_addr: TransportAddr,
         peer_identity: PeerIdentity,
+    ) -> Result<(), NodeError> {
+        self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?;
+        self.send_stored_msg1(link_id, transport_id, &remote_addr)
+            .await;
+        Ok(())
+    }
+
+    /// Prepare an outbound Noise msg1 at dial: allocate the session index, run
+    /// the Noise leaf, frame the wire, arm the shell-side resend, track
+    /// `pending_outbound`, and persist the connection + control machine (parked
+    /// in `Discovered`). Returns `Err` on index-allocation or Noise failure
+    /// (cleaning the partial leg), leaving the armed wire on the connection for
+    /// `send_stored_msg1` to transmit. Does NOT send — so the fallible setup can
+    /// propagate its error synchronously before any machine drive.
+    pub(in crate::node) fn prepare_outbound_msg1(
+        &mut self,
+        link_id: LinkId,
+        transport_id: TransportId,
+        remote_addr: &TransportAddr,
+        peer_identity: PeerIdentity,
     ) -> Result<(), NodeError> {
         let peer_node_addr = *peer_identity.node_addr();
 
@@ -538,7 +588,8 @@ impl Node {
             Err(e) => {
                 // Clean up the link we just created
                 self.links.remove(&link_id);
-                self.addr_to_link.remove(&(transport_id, remote_addr));
+                self.addr_to_link
+                    .remove(&(transport_id, remote_addr.clone()));
                 return Err(NodeError::IndexAllocationFailed(e.to_string()));
             }
         };
@@ -552,7 +603,8 @@ impl Node {
                     // Clean up the index and link
                     let _ = self.index_allocator.free(our_index);
                     self.links.remove(&link_id);
-                    self.addr_to_link.remove(&(transport_id, remote_addr));
+                    self.addr_to_link
+                        .remove(&(transport_id, remote_addr.clone()));
                     return Err(NodeError::HandshakeFailed(e.to_string()));
                 }
             };
@@ -576,7 +628,7 @@ impl Node {
 
         // Store msg1 for resend and schedule first resend
         let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms;
-        connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval);
+        connection.set_handshake_msg1(wire_msg1, current_time_ms + resend_interval);
 
         // Track in pending_outbound for msg2 dispatch
         self.pending_outbound
@@ -586,7 +638,8 @@ impl Node {
         // Persist the outbound control machine at dial, keyed by the same
         // `link_id` as the connection. It parks in `Discovered` (inert to reap
         // and rekey — it is absent from `peers`, never established) until
-        // `handle_msg2` looks it up to drive the promote. Its `our_index` is
+        // `handle_msg2` looks it up to drive the promote, or a connectionless
+        // dial drives it to `Handshaking` to send. Its `our_index` is
         // deliberately left unset so a later inbound restart does not emit a
         // spurious `UnregisterDecryptSession`. It is removed wherever the
         // connection is torn down without promoting (the stale reaper and the
@@ -595,16 +648,42 @@ impl Node {
         let machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms);
         self.peer_machines.insert(link_id, machine);
 
+        Ok(())
+    }
+
+    /// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection.
+    /// On send error, marks the connection failed and RETAINS it (the legacy
+    /// resend tick retries); a missing wire or transport is a no-op. This is the
+    /// body of the executor's `SendHandshake` msg1 action and the send tail of
+    /// the connection-oriented `start_handshake`.
+    pub(in crate::node) async fn send_stored_msg1(
+        &mut self,
+        link_id: LinkId,
+        transport_id: TransportId,
+        remote_addr: &TransportAddr,
+    ) {
+        let wire_msg1 = match self
+            .connections
+            .get(&link_id)
+            .and_then(|c| c.handshake_msg1())
+        {
+            Some(w) => w.to_vec(),
+            None => return,
+        };
+        let our_index = self.connections.get(&link_id).and_then(|c| c.our_index());
+
         // Send the wire format handshake message
         if let Some(transport) = self.transports.get(&transport_id) {
-            match transport.send(&remote_addr, &wire_msg1).await {
+            match transport.send(remote_addr, &wire_msg1).await {
                 Ok(bytes) => {
-                    debug!(
-                        link_id = %link_id,
-                        our_index = %our_index,
-                        bytes,
-                        "Sent Noise handshake message 1 (wire format)"
-                    );
+                    if let Some(idx) = our_index {
+                        debug!(
+                            link_id = %link_id,
+                            our_index = %idx,
+                            bytes,
+                            "Sent Noise handshake message 1 (wire format)"
+                        );
+                    }
                 }
                 Err(e) => {
                     warn!(
@@ -620,8 +699,6 @@ impl Node {
                 }
             }
         }
-
-        Ok(())
     }
 
     /// Poll all transports for discovered peers and auto-connect.
diff --git a/src/peer/machine.rs b/src/peer/machine.rs
index e4f0ba8..02e1990 100644
--- a/src/peer/machine.rs
+++ b/src/peer/machine.rs
@@ -188,11 +188,15 @@ pub(crate) enum TimerKind {
 ///
 /// Not `Debug`/`PartialEq`: the reused core snapshot payloads derive neither.
 pub(crate) enum PeerEvent {
-    /// Reconciler dial intent.
+    /// Reconciler dial intent. `connection_oriented` selects the outbound
+    /// path: connection-oriented transports open the transport first
+    /// (`OpenTransport` → `Connecting`); connectionless ones send msg1
+    /// immediately (`start_outbound_handshake` → `Handshaking`).
     Dial {
         transport_id: TransportId,
         remote_addr: TransportAddr,
         peer_identity: PeerIdentity,
+        connection_oriented: bool,
     },
     /// Connection-oriented transport connected.
     TransportConnected,
@@ -459,8 +463,9 @@ impl PeerMachine {
             PeerEvent::Dial {
                 transport_id,
                 remote_addr,
+                connection_oriented,
                 ..
-            } => self.on_dial(transport_id, remote_addr, now),
+            } => self.on_dial(transport_id, remote_addr, connection_oriented, now),
             PeerEvent::TransportConnected => self.on_transport_connected(now),
             PeerEvent::TransportFailed => self.on_transport_failed(now),
             PeerEvent::InboundMsg1 { link, wire, est } => {
@@ -513,21 +518,28 @@ impl PeerMachine {
         &mut self,
         transport_id: TransportId,
         remote_addr: TransportAddr,
-        _now: u64,
+        connection_oriented: bool,
+        now: u64,
     ) -> Vec<PeerAction> {
         if !matches!(self.state, PeerState::Discovered) {
             return Vec::new();
         }
         self.conn.set_transport_id(transport_id);
-        // Connection-oriented transports open the transport first; connectionless
-        // ones send msg1 immediately. This models the connection-oriented arm
-        // (OpenTransport) — the reconciler's candidate carries the transport
-        // kind at wiring time; connectionless dial reuses `start_handshake`.
-        self.state = PeerState::Connecting { link: self.link };
-        vec![PeerAction::OpenTransport {
-            transport_id,
-            remote_addr,
-        }]
+        if connection_oriented {
+            // Connection-oriented transports open the transport first; the
+            // executor's `OpenTransport` arm connects, then feeds
+            // `TransportConnected` → `start_outbound_handshake`.
+            self.state = PeerState::Connecting { link: self.link };
+            vec![PeerAction::OpenTransport {
+                transport_id,
+                remote_addr,
+            }]
+        } else {
+            // Connectionless transports have no connect step — send msg1
+            // immediately (the executor's `SendHandshake` msg1 branch performs
+            // the Noise leaf, framing, index alloc, and send).
+            self.start_outbound_handshake(now)
+        }
     }
 
     fn on_transport_connected(&mut self, now: u64) -> Vec<PeerAction> {
@@ -2093,6 +2105,69 @@ mod tests {
         );
     }
 
+    // ---- Test 7c: connectionless dial reaches Handshaking, Msg2 neutral ----
+    // The connectionless cutover drives the outbound machine
+    // Discovered -> (Dial, connection_oriented=false) -> Handshaking{SentMsg1}
+    // BEFORE msg2, whereas the pre-cutover path stepped Msg2 while still in
+    // Discovered. `on_msg2` is state-independent, so both must yield the
+    // identical `[PromoteToActive]` and leave `our_index == None`.
+    #[test]
+    fn connectionless_dial_then_msg2_promotes_from_handshaking() {
+        let mut alloc = IndexAllocator::new();
+        let peer = peer_identity();
+
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0);
+        // Connectionless dial: no OpenTransport, straight to Handshaking{SentMsg1}.
+        let dial = m.step(
+            PeerEvent::Dial {
+                transport_id: TransportId::new(1),
+                remote_addr: TransportAddr::from_string("127.0.0.1:9999"),
+                peer_identity: peer,
+                connection_oriented: false,
+            },
+            100,
+            &mut alloc,
+        );
+        assert!(matches!(
+            m.state(),
+            PeerState::Handshaking {
+                phase: HandshakePhase::SentMsg1,
+                ..
+            }
+        ));
+        assert!(
+            dial.iter()
+                .any(|a| matches!(a, PeerAction::SendHandshake { .. }))
+        );
+        assert!(
+            !dial
+                .iter()
+                .any(|a| matches!(a, PeerAction::OpenTransport { .. })),
+            "connectionless dial emits no OpenTransport"
+        );
+
+        // Step Msg2 from Handshaking — identical promote to the Discovered path.
+        let out = OutboundSnapshot {
+            has_existing_peer: false,
+            our_outbound_wins: false,
+        };
+        let promote = m.step(
+            PeerEvent::Msg2 {
+                their_index: SessionIndex::new(0x77),
+                out,
+            },
+            200,
+            &mut alloc,
+        );
+        assert_eq!(
+            promote,
+            vec![PeerAction::PromoteToActive {
+                link: LinkId::new(1)
+            }]
+        );
+        assert_eq!(m.our_index(), None);
+    }
+
     // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost --------------
     #[test]
     fn liveness_to_link_dead() {

From f698da50b65e2efc33297bb173afc322f138ed7b Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 13:52:34 +0000
Subject: [PATCH 4/6] peer: add unit coverage for the connection-oriented
 outbound dial path

The connection-oriented (TCP/Tor) outbound connect->handshake path had no
`cargo test --lib` coverage; it was exercised only by the opt-in Tor
integration suites, and the TCP node tests bypass it via a manual
connectionless handshake helper. Add three unit tests:

- a machine-level test driving Dial{connection_oriented:true} -> Connecting
  (emitting only OpenTransport, no msg1) -> TransportConnected ->
  Handshaking{SentMsg1}, asserting the exact OpenTransport and
  SendHandshake+SetTimer action vectors that start_outbound_handshake emits
  (its oriented reach via on_transport_connected was previously untested; the
  connectionless reach via on_dial was already covered);

- two node-level tests over a real loopback TcpTransport: a successful connect
  that reaches start_handshake (observed via a pending_outbound entry), and a
  connect to a closed port that routes through the failure arm (link torn down,
  no msg1 dispatched).

Tests only; no production change.
---
 src/node/tests/tcp.rs | 110 ++++++++++++++++++++++++++++++++++++++++++
 src/peer/machine.rs   |  65 +++++++++++++++++++++++++
 2 files changed, 175 insertions(+)

diff --git a/src/node/tests/tcp.rs b/src/node/tests/tcp.rs
index f120afa..393fb62 100644
--- a/src/node/tests/tcp.rs
+++ b/src/node/tests/tcp.rs
@@ -286,3 +286,113 @@ async fn test_tcp_reconnection_after_link_death() {
 
     cleanup_nodes(&mut nodes).await;
 }
+
+/// Connection-oriented outbound connect succeeds on loopback and reaches the
+/// handshake.
+///
+/// `initiate_connection` opens a real non-blocking TCP connect to a live
+/// listener (node 1's `TcpTransport`) and queues a `PendingConnect`;
+/// `poll_pending_connects` promotes the resolved connection to `Connected` and
+/// runs `start_handshake`. The observable is `pending_outbound`: `start_handshake`
+/// → `prepare_outbound_msg1` tracks the dispatched msg1 there (keyed by the
+/// allocated session index), which is empty until the connect resolves.
+#[tokio::test]
+async fn test_tcp_oriented_connect_success_reaches_handshake() {
+    let mut nodes = vec![make_test_node_tcp().await, make_test_node_tcp().await];
+
+    // Target: node 1's live TCP listener + its verified identity.
+    let target_addr = nodes[1].addr.clone();
+    let target_identity = PeerIdentity::from_pubkey_full(nodes[1].node.identity().pubkey_full());
+    let transport_id = nodes[0].transport_id;
+
+    // Kick off the non-blocking oriented connect.
+    nodes[0]
+        .node
+        .initiate_connection(transport_id, target_addr, target_identity)
+        .await
+        .expect("initiate_connection should queue a pending connect");
+
+    // One pending connect queued, no msg1 dispatched yet.
+    assert_eq!(nodes[0].node.peering.pending_connects.len(), 1);
+    assert!(nodes[0].node.pending_outbound.is_empty());
+
+    // Drive the tick-side poll until the background connect resolves.
+    for _ in 0..100 {
+        nodes[0].node.poll_pending_connects().await;
+        if nodes[0].node.peering.pending_connects.is_empty() {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    assert!(
+        nodes[0].node.peering.pending_connects.is_empty(),
+        "connect should have resolved"
+    );
+    // Reaching start_handshake -> prepare_outbound_msg1 tracks the outbound msg1
+    // in pending_outbound: proof the connect-success path dispatched the handshake.
+    assert_eq!(
+        nodes[0].node.pending_outbound.len(),
+        1,
+        "connect-success path should have reached start_handshake and dispatched msg1"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
+
+/// Connection-oriented outbound connect to a closed port fails and tears down
+/// the link.
+///
+/// The dial targets a guaranteed-closed loopback port (bind then drop the
+/// listener). `initiate_connection` is non-blocking, so it still queues a
+/// `PendingConnect` and a `Connecting` link; `poll_pending_connects` observes the
+/// background task's `Failed` result and routes through the failure arm
+/// (`remove_link` + `note_handshake_timeout`). The observable is the removed link
+/// and the absent `pending_outbound` entry — the failure path never reaches
+/// `start_handshake`.
+#[tokio::test]
+async fn test_tcp_oriented_connect_failure_tears_down_link() {
+    let mut nodes = vec![make_test_node_tcp().await];
+
+    // A guaranteed-closed loopback port: bind to grab a port, then drop it.
+    let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let closed_addr = closed.local_addr().unwrap();
+    drop(closed);
+    let target_addr = TransportAddr::from_string(&closed_addr.to_string());
+    let target_identity = make_peer_identity();
+    let transport_id = nodes[0].transport_id;
+
+    nodes[0]
+        .node
+        .initiate_connection(transport_id, target_addr, target_identity)
+        .await
+        .expect("initiate_connection is non-blocking; it queues a pending connect");
+
+    // One pending connect + the in-flight dial's link.
+    assert_eq!(nodes[0].node.peering.pending_connects.len(), 1);
+    assert_eq!(nodes[0].node.links.len(), 1);
+
+    for _ in 0..100 {
+        nodes[0].node.poll_pending_connects().await;
+        if nodes[0].node.peering.pending_connects.is_empty() {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    assert!(
+        nodes[0].node.peering.pending_connects.is_empty(),
+        "failed connect should have been drained"
+    );
+    // Failure path removes the link and never reaches start_handshake.
+    assert!(
+        nodes[0].node.links.is_empty(),
+        "connect-failure path should have torn down the link"
+    );
+    assert!(
+        nodes[0].node.pending_outbound.is_empty(),
+        "connect-failure path must not dispatch msg1"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
diff --git a/src/peer/machine.rs b/src/peer/machine.rs
index 02e1990..5305400 100644
--- a/src/peer/machine.rs
+++ b/src/peer/machine.rs
@@ -2168,6 +2168,71 @@ mod tests {
         assert_eq!(m.our_index(), None);
     }
 
+    // ---- Test 7d: connection-oriented dial opens transport first ----------
+    // The connection-oriented cutover drives the outbound machine
+    // Discovered -> (Dial, connection_oriented=true) -> Connecting
+    // (emitting ONLY OpenTransport, no msg1 yet), then TransportConnected ->
+    // Handshaking{SentMsg1} with the same SendHandshake + two SetTimer that
+    // `start_outbound_handshake` emits. Covers the oriented reach into
+    // `start_outbound_handshake` via `on_transport_connected` (the connectionless
+    // reach via `on_dial` is already covered by the test above).
+    #[test]
+    fn connection_oriented_dial_opens_transport_then_connected_handshakes() {
+        let mut alloc = IndexAllocator::new();
+        let peer = peer_identity();
+
+        let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0);
+        // Connection-oriented dial: open the transport first, no msg1 yet.
+        let dial = m.step(
+            PeerEvent::Dial {
+                transport_id: TransportId::new(1),
+                remote_addr: TransportAddr::from_string("127.0.0.1:9999"),
+                peer_identity: peer,
+                connection_oriented: true,
+            },
+            100,
+            &mut alloc,
+        );
+        assert_eq!(
+            m.state(),
+            PeerState::Connecting {
+                link: LinkId::new(1)
+            }
+        );
+        assert_eq!(
+            dial,
+            vec![PeerAction::OpenTransport {
+                transport_id: TransportId::new(1),
+                remote_addr: TransportAddr::from_string("127.0.0.1:9999"),
+            }],
+            "connection-oriented dial emits exactly one OpenTransport and no msg1"
+        );
+
+        // Transport connected: now send msg1 and arm the handshake timers.
+        let connected = m.step(PeerEvent::TransportConnected, 200, &mut alloc);
+        assert!(matches!(
+            m.state(),
+            PeerState::Handshaking {
+                phase: HandshakePhase::SentMsg1,
+                ..
+            }
+        ));
+        assert_eq!(
+            connected,
+            vec![
+                PeerAction::SendHandshake { bytes: Vec::new() },
+                PeerAction::SetTimer {
+                    kind: TimerKind::HandshakeRetransmit,
+                    at_ms: 200 + HANDSHAKE_RETRANSMIT_INTERVAL_MS,
+                },
+                PeerAction::SetTimer {
+                    kind: TimerKind::HandshakeTimeout,
+                    at_ms: 200 + HANDSHAKE_TIMEOUT_MS,
+                },
+            ]
+        );
+    }
+
     // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost --------------
     #[test]
     fn liveness_to_link_dead() {

From 9588c500638ba15224c3b1405d8c4bc217470edd Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 14:12:49 +0000
Subject: [PATCH 5/6] peer: persist the outbound control machine at dial, not
 at msg1 prepare

Move the peer_machines insert for an outbound leg out of the tail of
prepare_outbound_msg1 to a single shared site in initiate_connection, before
the connection-oriented / connectionless fork. This lets the connection-oriented
path find the machine at dial time (so it can be driven through the connect
handshake) without prepare_outbound_msg1 -- which for that path runs after the
connect completes -- clobbering an in-progress machine back to Discovered.

Because the machine now exists before the fallible dial steps, add
peer_machines.remove to every failure path in the widened dial window: the
index-allocation and Noise-leaf failures in prepare_outbound_msg1, the oriented
transport.connect failure, and both poll_pending_connects teardown arms
(handshake-start failure and async connect failure). remove_link does not touch
peer_machines, so these explicit removes are required.

Behavior-neutral: the connectionless path still drives the machine to
Handshaking and sends msg1 identically; the machine's our_index stays unset
(no spurious UnregisterDecryptSession on a later inbound restart); a failed dial
leaves peer_machines empty for that link exactly as before; and no
connection-oriented machine drive is wired here.
---
 src/node/lifecycle/mod.rs | 53 ++++++++++++++++++++++-----------------
 1 file changed, 30 insertions(+), 23 deletions(-)

diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs
index 2f32a14..927f465 100644
--- a/src/node/lifecycle/mod.rs
+++ b/src/node/lifecycle/mod.rs
@@ -481,6 +481,18 @@ impl Node {
         self.addr_to_link
             .insert((transport_id, remote_addr.clone()), link_id);
 
+        // Persist the outbound control machine at dial, keyed by the same
+        // `link_id` as the (soon-to-be-built) connection. It parks in
+        // `Discovered` (inert to reap and rekey — absent from `peers`, never
+        // established) until `handle_msg2` looks it up to drive the promote, or
+        // a connectionless dial drives it to `Handshaking` to send. Its
+        // `our_index` is deliberately left unset so a later inbound restart does
+        // not emit a spurious `UnregisterDecryptSession`. It is removed on every
+        // failure path in the dial window (below and in `prepare_outbound_msg1`
+        // / `poll_pending_connects`), mirroring the connection's own lifetime.
+        let machine = PeerMachine::new_outbound(link_id, peer_identity, Self::now_ms());
+        self.peer_machines.insert(link_id, machine);
+
         if is_connection_oriented {
             // Connection-oriented: start non-blocking connect, defer handshake
             if let Some(transport) = self.transports.get(&transport_id) {
@@ -501,9 +513,10 @@ impl Node {
                         });
                     }
                     Err(e) => {
-                        // Clean up link
+                        // Clean up link and the dial-time control machine
                         self.links.remove(&link_id);
                         self.addr_to_link.remove(&(transport_id, remote_addr));
+                        self.peer_machines.remove(&link_id);
                         return Err(NodeError::TransportError(e.to_string()));
                     }
                 }
@@ -564,11 +577,14 @@ impl Node {
 
     /// Prepare an outbound Noise msg1 at dial: allocate the session index, run
     /// the Noise leaf, frame the wire, arm the shell-side resend, track
-    /// `pending_outbound`, and persist the connection + control machine (parked
-    /// in `Discovered`). Returns `Err` on index-allocation or Noise failure
-    /// (cleaning the partial leg), leaving the armed wire on the connection for
-    /// `send_stored_msg1` to transmit. Does NOT send — so the fallible setup can
-    /// propagate its error synchronously before any machine drive.
+    /// `pending_outbound`, and persist the connection. Returns `Err` on
+    /// index-allocation or Noise failure (cleaning the partial leg, including
+    /// the dial-time control machine), leaving the armed wire on the connection
+    /// for `send_stored_msg1` to transmit. Does NOT send — so the fallible setup
+    /// can propagate its error synchronously before any machine drive. The
+    /// control machine itself is persisted at dial in `initiate_connection`;
+    /// this function no longer touches `peer_machines` except to clean it up on
+    /// the failure paths.
     pub(in crate::node) fn prepare_outbound_msg1(
         &mut self,
         link_id: LinkId,
@@ -586,10 +602,11 @@ impl Node {
         let our_index = match self.index_allocator.allocate() {
             Ok(idx) => idx,
             Err(e) => {
-                // Clean up the link we just created
+                // Clean up the link and dial-time machine we just created
                 self.links.remove(&link_id);
                 self.addr_to_link
                     .remove(&(transport_id, remote_addr.clone()));
+                self.peer_machines.remove(&link_id);
                 return Err(NodeError::IndexAllocationFailed(e.to_string()));
             }
         };
@@ -600,11 +617,12 @@ impl Node {
             match connection.start_handshake(our_keypair, self.startup_epoch(), current_time_ms) {
                 Ok(msg) => msg,
                 Err(e) => {
-                    // Clean up the index and link
+                    // Clean up the index, link, and dial-time machine
                     let _ = self.index_allocator.free(our_index);
                     self.links.remove(&link_id);
                     self.addr_to_link
                         .remove(&(transport_id, remote_addr.clone()));
+                    self.peer_machines.remove(&link_id);
                     return Err(NodeError::HandshakeFailed(e.to_string()));
                 }
             };
@@ -635,19 +653,6 @@ impl Node {
             .insert((transport_id, our_index.as_u32()), link_id);
         self.connections.insert(link_id, connection);
 
-        // Persist the outbound control machine at dial, keyed by the same
-        // `link_id` as the connection. It parks in `Discovered` (inert to reap
-        // and rekey — it is absent from `peers`, never established) until
-        // `handle_msg2` looks it up to drive the promote, or a connectionless
-        // dial drives it to `Handshaking` to send. Its `our_index` is
-        // deliberately left unset so a later inbound restart does not emit a
-        // spurious `UnregisterDecryptSession`. It is removed wherever the
-        // connection is torn down without promoting (the stale reaper and the
-        // msg2 ACL-fail / cross-connection arms), mirroring the connection's
-        // own lifetime.
-        let machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms);
-        self.peer_machines.insert(link_id, machine);
-
         Ok(())
     }
 
@@ -1233,8 +1238,9 @@ impl Node {
                         error = %e,
                         "Failed to start handshake after transport connect"
                     );
-                    // Clean up link on handshake failure
+                    // Clean up link and dial-time machine on handshake failure
                     self.remove_link(&pending.link_id);
+                    self.peer_machines.remove(&pending.link_id);
                 }
             } else {
                 let reason = reason.unwrap_or_default();
@@ -1247,9 +1253,10 @@ impl Node {
                     "Transport connect failed"
                 );
 
-                // Clean up link and schedule retry
+                // Clean up link and dial-time machine, then schedule retry
                 self.remove_link(&pending.link_id);
                 self.links.remove(&pending.link_id);
+                self.peer_machines.remove(&pending.link_id);
                 self.note_handshake_timeout(*pending.peer_identity.node_addr(), Self::now_ms());
             }
         }

From 3e7ca90212e6473c2e93b108d695c8bfe09b4bc8 Mon Sep 17 00:00:00 2001
From: Johnathan Corgan <johnathan@corganlabs.com>
Date: Wed, 15 Jul 2026 14:40:58 +0000
Subject: [PATCH 6/6] peer: drive the connection-oriented outbound dial through
 the machine

Route the connection-oriented (TCP/Tor) outbound path through the peer state
machine, matching how the connectionless path already works.
initiate_connection's oriented branch now drives PeerEvent::Dial with
connection_oriented=true; the machine parks in Connecting and emits
OpenTransport, whose executor arm performs the non-blocking transport.connect
and pushes the PendingConnect. When the connect resolves, poll_pending_connects
prepares msg1 in the shell and then drives PeerEvent::TransportConnected, which
sends msg1 via the machine's SendHandshake arm.

The msg1 prepare (index allocation, Noise leaf, wire arming) MUST run in the
shell before the TransportConnected drive: send_stored_msg1 only transmits an
already-armed wire, so a drive-only path would silently send nothing. The
machine's our_index stays unset; the connect-failure path keeps its direct
handshake-timeout handling (TransportFailed stays dormant). The now-unused
Node::start_handshake helper is removed.

Behavior-neutral: same transport.connect, same PendingConnect, same msg1 send
and failure teardown as the removed inline path -- only the driver changes from
inline shell code to the state machine.
---
 src/node/dataplane/peer_actions.rs |  59 +++++++++++---
 src/node/handlers/handshake.rs     |   8 +-
 src/node/lifecycle/mod.rs          | 126 +++++++++++++++--------------
 3 files changed, 119 insertions(+), 74 deletions(-)

diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs
index 74b99e8..8dd2eaf 100644
--- a/src/node/dataplane/peer_actions.rs
+++ b/src/node/dataplane/peer_actions.rs
@@ -10,15 +10,17 @@
 //!
 //! The executor is wired incrementally. Live today: the inbound establish
 //! (`handle_msg1` → `step(InboundMsg1)`), the outbound msg2 promote
-//! (`handle_msg2` looks up the dial-persisted machine), and the connectionless
+//! (`handle_msg2` looks up the dial-persisted machine), the connectionless
 //! outbound msg1 send (`SendHandshake` with `their_index == None` →
-//! `send_stored_msg1`, driven from `initiate_connection`).
+//! `send_stored_msg1`, driven from `initiate_connection`), and the
+//! connection-oriented dial (`OpenTransport` performs the non-blocking
+//! `transport.connect`; `TransportConnected` drives the connect-resolution msg1
+//! send from `poll_pending_connects`).
 //!
-//! Not yet driven, so their arms stay inert stubs: `OpenTransport` (the
-//! connection-oriented dial), 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.
+//! 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.
 
 use crate::PeerIdentity;
 use crate::node::Node;
@@ -109,10 +111,45 @@ impl Node {
         let mut queue: VecDeque<PeerAction> = actions.into();
         while let Some(action) = queue.pop_front() {
             match action {
-                PeerAction::OpenTransport { .. } => {
-                    // Outbound dial (`initiate_connection`,
-                    // `lifecycle/mod.rs:470`). Outbound establish is not cut over
-                    // yet; inert in the shadow-only skeleton.
+                PeerAction::OpenTransport {
+                    transport_id,
+                    remote_addr,
+                } => {
+                    // Outbound connection-oriented dial. `initiate_connection`'s
+                    // oriented branch drove the machine to `Connecting`, which
+                    // emitted this action. Perform the non-blocking
+                    // `transport.connect` and, on success, push the
+                    // `PendingConnect` for `poll_pending_connects` to resolve. On
+                    // connect error, tear down the dial-window state (link,
+                    // reverse map, control machine) and abort the queue — the
+                    // executor-local mirror of the old inline
+                    // `initiate_connection` connect+push.
+                    if let Some(transport) = self.transports.get(&transport_id) {
+                        match transport.connect(&remote_addr).await {
+                            Ok(()) => {
+                                debug!(
+                                    transport_id = %transport_id,
+                                    remote_addr = %remote_addr,
+                                    link_id = %link,
+                                    "Transport connect initiated (non-blocking)"
+                                );
+                                self.peering
+                                    .pending_connects
+                                    .push(crate::node::PendingConnect {
+                                        link_id: link,
+                                        transport_id,
+                                        remote_addr,
+                                        peer_identity: ambient.verified_identity,
+                                    });
+                            }
+                            Err(_e) => {
+                                self.links.remove(&link);
+                                self.addr_to_link.remove(&(transport_id, remote_addr));
+                                self.peer_machines.remove(&link);
+                                return;
+                            }
+                        }
+                    }
                 }
                 PeerAction::SendHandshake { bytes } => {
                     // Two outbound directions share this action, discriminated by
diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs
index 3f78903..87563ab 100644
--- a/src/node/handlers/handshake.rs
+++ b/src/node/handlers/handshake.rs
@@ -1141,10 +1141,10 @@ impl Node {
         // net-new arm — no ordering constraint, lowest risk.
         //
         // Look up the outbound machine persisted at DIAL, and step `Msg2 →
-        // [PromoteToActive]` in place. It is in `Discovered` (connection-oriented
-        // dial, not yet cut over) or `Handshaking{SentMsg1}` (connectionless dial,
-        // which drives the machine to send msg1) — either way `on_msg2` is
-        // state-independent: it decides from the outbound snapshot, not the
+        // [PromoteToActive]` in place. Both dial paths reach msg2 in
+        // `Handshaking{SentMsg1}` — each drives the machine to send msg1 before
+        // msg2 — and `on_msg2` is state-independent regardless: it decides from
+        // the outbound snapshot, not the
         // machine's state, and reads the machine's unset `conn.our_index`
         // as `None`, so stepping the persisted (vs the former transient) machine
         // is byte-identical: same `Promote` decision (`has_existing_peer == false`),
diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs
index 927f465..ec1c683 100644
--- a/src/node/lifecycle/mod.rs
+++ b/src/node/lifecycle/mod.rs
@@ -439,8 +439,6 @@ impl Node {
         remote_addr: TransportAddr,
         peer_identity: PeerIdentity,
     ) -> Result<(), NodeError> {
-        let peer_node_addr = *peer_identity.node_addr();
-
         self.authorize_peer(
             &peer_identity,
             PeerAclContext::OutboundConnect,
@@ -494,33 +492,35 @@ impl Node {
         self.peer_machines.insert(link_id, machine);
 
         if is_connection_oriented {
-            // Connection-oriented: start non-blocking connect, defer handshake
-            if let Some(transport) = self.transports.get(&transport_id) {
-                match transport.connect(&remote_addr).await {
-                    Ok(()) => {
-                        debug!(
-                            peer = %self.peer_display_name(&peer_node_addr),
-                            transport_id = %transport_id,
-                            remote_addr = %remote_addr,
-                            link_id = %link_id,
-                            "Transport connect initiated (non-blocking)"
-                        );
-                        self.peering.pending_connects.push(super::PendingConnect {
-                            link_id,
-                            transport_id,
-                            remote_addr,
-                            peer_identity,
-                        });
-                    }
-                    Err(e) => {
-                        // Clean up link and the dial-time control machine
-                        self.links.remove(&link_id);
-                        self.addr_to_link.remove(&(transport_id, remote_addr));
-                        self.peer_machines.remove(&link_id);
-                        return Err(NodeError::TransportError(e.to_string()));
-                    }
-                }
-            }
+            // Connection-oriented: drive the machine to open the transport. The
+            // machine parks in `Connecting` and emits one `OpenTransport`; the
+            // executor performs the non-blocking `transport.connect` and pushes
+            // the `PendingConnect`. No index alloc or wire-arm happens here —
+            // that is deferred to `prepare_outbound_msg1` at connect-resolution
+            // time (`poll_pending_connects`), so `our_index` stays `None` on both
+            // the (not-yet-built) connection and the machine.
+            let now = Self::now_ms();
+            let ambient = PeerActionCtx {
+                verified_identity: peer_identity,
+                transport_id,
+                remote_addr: remote_addr.clone(),
+                our_index: None,
+                their_index: None,
+                now_ms: now,
+                is_outbound: true,
+            };
+            self.advance_peer_machine(
+                link_id,
+                PeerEvent::Dial {
+                    transport_id,
+                    remote_addr,
+                    peer_identity,
+                    connection_oriented: true,
+                },
+                now,
+                &ambient,
+            )
+            .await;
             Ok(())
         } else {
             // Connectionless: no connect step. Prepare msg1 in the shell — the
@@ -557,24 +557,6 @@ impl Node {
         }
     }
 
-    /// Start the Noise handshake on a link and send msg1.
-    ///
-    /// Called after a connection-oriented transport connects. (Connectionless
-    /// dials `prepare_outbound_msg1` + drive the machine to send in
-    /// `initiate_connection`.)
-    pub(super) async fn start_handshake(
-        &mut self,
-        link_id: LinkId,
-        transport_id: TransportId,
-        remote_addr: TransportAddr,
-        peer_identity: PeerIdentity,
-    ) -> Result<(), NodeError> {
-        self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?;
-        self.send_stored_msg1(link_id, transport_id, &remote_addr)
-            .await;
-        Ok(())
-    }
-
     /// Prepare an outbound Noise msg1 at dial: allocate the session index, run
     /// the Noise leaf, frame the wire, arm the shell-side resend, track
     /// `pending_outbound`, and persist the connection. Returns `Err` on
@@ -659,8 +641,9 @@ impl Node {
     /// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection.
     /// On send error, marks the connection failed and RETAINS it (the legacy
     /// resend tick retries); a missing wire or transport is a no-op. This is the
-    /// body of the executor's `SendHandshake` msg1 action and the send tail of
-    /// the connection-oriented `start_handshake`.
+    /// body of the executor's `SendHandshake` msg1 action — reached on both the
+    /// connectionless dial and the connection-oriented connect-resolution path,
+    /// after `prepare_outbound_msg1` has armed the wire.
     pub(in crate::node) async fn send_stored_msg1(
         &mut self,
         link_id: LinkId,
@@ -1223,16 +1206,18 @@ impl Node {
                     "Transport connected, starting handshake"
                 );
 
-                // Start the handshake now that the transport is connected
-                if let Err(e) = self
-                    .start_handshake(
-                        pending.link_id,
-                        pending.transport_id,
-                        pending.remote_addr.clone(),
-                        pending.peer_identity,
-                    )
-                    .await
-                {
+                // Prepare msg1 now that the transport is connected, then drive
+                // the machine to send it. The prepare (index alloc, Noise leaf,
+                // wire arm) MUST run BEFORE the `TransportConnected` drive: the
+                // executor's `SendHandshake` msg1 branch only transmits the wire
+                // this armed on the connection — a drive-only path would find no
+                // armed wire and silently send nothing.
+                if let Err(e) = self.prepare_outbound_msg1(
+                    pending.link_id,
+                    pending.transport_id,
+                    &pending.remote_addr,
+                    pending.peer_identity,
+                ) {
                     warn!(
                         link_id = %pending.link_id,
                         error = %e,
@@ -1241,6 +1226,29 @@ 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);
+                } else {
+                    // Drive the dial-persisted machine: `Connecting` →
+                    // `on_transport_connected` → `start_outbound_handshake`,
+                    // emitting `SendHandshake` (msg1) whose executor arm sends the
+                    // wire just armed. `our_index`/`their_index` stay `None` so the
+                    // executor takes the `their_index == None` msg1 branch.
+                    let now = Self::now_ms();
+                    let ambient = PeerActionCtx {
+                        verified_identity: pending.peer_identity,
+                        transport_id: pending.transport_id,
+                        remote_addr: pending.remote_addr.clone(),
+                        our_index: None,
+                        their_index: None,
+                        now_ms: now,
+                        is_outbound: true,
+                    };
+                    self.advance_peer_machine(
+                        pending.link_id,
+                        PeerEvent::TransportConnected,
+                        now,
+                        &ambient,
+                    )
+                    .await;
                 }
             } else {
                 let reason = reason.unwrap_or_default();