diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go
new file mode 100644
index 00000000..fadd40ef
--- /dev/null
+++ b/cmd/pilotctl/firstcontact.go
@@ -0,0 +1,365 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+// First-contact request/reply for send-message --wait.
+//
+// A fresh node's first query to a service (the onboarding step
+// `pilotctl send-message list-agents --data ... --wait`) has to build the
+// whole path inside one command: resolve the peer, key exchange, SYN,
+// the request, then the service's reply, which comes back on a NEW
+// connection the service dials to our port 1001. Measured on clean
+// GitHub runners the first query failed most of the time, for reasons
+// that stack:
+//
+//   - the reply SYN was dropped by our own private-node trust gate until
+//     a trust handshake with the service finished (fixed in the daemon:
+//     reply window, replywindow.go);
+//   - pilotctl blocked up to ~22 s on that handshake before it even
+//     dialed, eating the dial budget;
+//   - the dial spent its SYN retries while the key exchange was still
+//     running (fixed in the daemon: dial_awaits_key);
+//   - the service answers once, from a responder with a sub-second send
+//     budget, so a reply lost while the service's side of the new path is
+//     still settling is never retried.
+//
+// This file holds the client half: detect what the daemon supports,
+// dial with one retry while the path is still converging, match the
+// reply to this request, re-send the request once on a new stream when a
+// first-contact reply does not show up, and say which step failed.
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/pilot-protocol/common/driver"
+	"github.com/pilot-protocol/common/protocol"
+)
+
+// daemonFeatureSet caches the "features" list of the daemon's info reply
+// for the life of this process (it only changes when the daemon does).
+var daemonFeatureSet map[string]bool
+
+// daemonHasFeature reports whether the local daemon advertises feature
+// (see daemonFeatures in pkg/daemon/ipc.go). An older daemon reports no
+// features, so every caller falls back to the behaviour it had before.
+func daemonHasFeature(d *driver.Driver, feature string) bool {
+	if daemonFeatureSet == nil {
+		daemonFeatureSet = map[string]bool{}
+		if info, err := d.Info(); err == nil {
+			for f := range featuresOf(info) {
+				daemonFeatureSet[f] = true
+			}
+		}
+	}
+	return daemonFeatureSet[feature]
+}
+
+func featuresOf(info map[string]interface{}) map[string]bool {
+	out := map[string]bool{}
+	list, _ := info["features"].([]interface{})
+	for _, v := range list {
+		if s, ok := v.(string); ok {
+			out[s] = true
+		}
+	}
+	return out
+}
+
+// peerSessionUp reports whether the daemon already holds an encrypted
+// session with node: if not, this command is making first contact.
+func peerSessionUp(info map[string]interface{}, node uint32) bool {
+	list, _ := info["peer_list"].([]interface{})
+	for _, v := range list {
+		p, ok := v.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		if id, ok := p["node_id"].(float64); ok && uint32(id) == node {
+			enc, _ := p["encrypted"].(bool)
+			return enc
+		}
+	}
+	return false
+}
+
+// replyConnsFrom counts inbound data-exchange connections from target in
+// the daemon's connection table: the service dialing its reply back.
+func replyConnsFrom(info map[string]interface{}, target string) int {
+	list, _ := info["conn_list"].([]interface{})
+	n := 0
+	for _, v := range list {
+		c, ok := v.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		lp, _ := c["local_port"].(float64)
+		ra, _ := c["remote_addr"].(string)
+		if uint16(lp) == protocol.PortDataExchange && ra == target {
+			n++
+		}
+	}
+	return n
+}
+
+// isConvergingDialError reports whether a failed dial is worth one more
+// try on first contact: the peer did not answer in time (key exchange
+// or SYN), as opposed to a refusal or a local error.
+func isConvergingDialError(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := err.Error()
+	return strings.Contains(s, "dial timeout") || strings.Contains(s, "key exchange")
+}
+
+// isDriverSideTimeout reports whether the dial error is the SDK giving up
+// on the IPC reply (the daemon may still be dialing), as opposed to an
+// error the daemon returned.
+func isDriverSideTimeout(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "dial timeout") && !strings.Contains(err.Error(), "daemon:")
+}
+
+// dialFailureHint explains a failed data-exchange dial.
+func dialFailureHint(target string, err error, attempts int, elapsed time.Duration) string {
+	s := ""
+	if err != nil {
+		s = err.Error()
+	}
+	switch {
+	case strings.Contains(s, "key exchange"):
+		return fmt.Sprintf("%s never completed the tunnel key exchange (%d dial attempt(s), %s): the service is overloaded or offline. "+
+			"Retry in a minute; `pilotctl peers` shows the tunnel state.", target, attempts, elapsed.Round(time.Second))
+	case strings.Contains(s, "dial timeout"):
+		return fmt.Sprintf("%s did not accept a connection on port %d (%d dial attempt(s), %s). "+
+			"It may be overloaded; retry in a minute or check reachability with `pilotctl ping %s`.",
+			target, protocol.PortDataExchange, attempts, elapsed.Round(time.Second), target)
+	}
+	if h := classifyDaemonError(err); h != "" {
+		return h
+	}
+	return fmt.Sprintf("check that %s is reachable: pilotctl ping %s", target, target)
+}
+
+// inboxWatch finds the reply to one request in the inbox. It snapshots
+// the files present before the request is sent, so a reply to an
+// earlier request (or the late duplicate some services send) that is
+// already sitting in the inbox is never taken for this one, and it
+// returns the oldest new file from the peer rather than whichever one
+// the directory listing happens to put first.
+type inboxWatch struct {
+	dir    string
+	from   string
+	cutoff time.Time
+	seen   map[string]bool
+}
+
+func inboxDirPath() (string, error) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(home, ".pilot", "inbox"), nil
+}
+
+// newInboxWatch snapshots the inbox now. from is the sender address the
+// daemon writes into each inbox record ("" matches any sender).
+func newInboxWatch(from string) *inboxWatch {
+	w := &inboxWatch{from: from, cutoff: time.Now().Add(-time.Second), seen: map[string]bool{}}
+	dir, err := inboxDirPath()
+	if err != nil {
+		return w
+	}
+	w.dir = dir
+	if entries, err := os.ReadDir(dir); err == nil {
+		for _, e := range entries {
+			w.seen[e.Name()] = true
+		}
+	}
+	return w
+}
+
+// poll returns the oldest reply that arrived since the snapshot.
+func (w *inboxWatch) poll() (map[string]interface{}, error) {
+	if w.dir == "" {
+		dir, err := inboxDirPath()
+		if err != nil {
+			return nil, err
+		}
+		w.dir = dir
+	}
+	entries, err := os.ReadDir(w.dir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("read inbox: %w", err)
+	}
+	// Inbox files are {TYPE}-{timestamp}-{seq}.json: order by the part
+	// after the type so the oldest new message wins.
+	sort.Slice(entries, func(i, j int) bool {
+		ni, nj := entries[i].Name(), entries[j].Name()
+		di, dj := strings.Index(ni, "-"), strings.Index(nj, "-")
+		if di < 0 || dj < 0 {
+			return ni < nj
+		}
+		return ni[di:] < nj[dj:]
+	})
+	for _, e := range entries {
+		if e.IsDir() || w.seen[e.Name()] {
+			continue
+		}
+		info, err := e.Info()
+		if err != nil || !info.ModTime().After(w.cutoff) {
+			continue
+		}
+		data, err := os.ReadFile(filepath.Join(w.dir, e.Name()))
+		if err != nil {
+			continue
+		}
+		var msg map[string]interface{}
+		if json.Unmarshal(data, &msg) != nil {
+			continue
+		}
+		if w.from != "" {
+			if from, _ := msg["from"].(string); from != w.from {
+				continue
+			}
+		}
+		return msg, nil
+	}
+	return nil, nil
+}
+
+// replyWait configures awaitReply.
+type replyWait struct {
+	// wait is the reply budget, counted from the moment the request was
+	// acknowledged by the receiver.
+	wait time.Duration
+	// resendAfter, when > 0, re-sends the request once if no reply has
+	// arrived that long after the acknowledgement.
+	resendAfter time.Duration
+	// resend sends the request again on a new stream and returns when the
+	// receiver acknowledged it. It runs on its own goroutine.
+	resend func() (ackAt time.Time, err error)
+	// observe returns how many reply connections from the peer the daemon
+	// currently holds (the service dialing back). Optional; polled ~1/s.
+	observe func() int
+}
+
+// replyOutcome is what awaitReply saw.
+type replyOutcome struct {
+	reply      map[string]interface{}
+	resent     bool
+	resendErr  error
+	replyConns int // most reply connections from the peer seen at once
+	waited     time.Duration
+}
+
+// awaitReply polls the inbox until the reply arrives or the budget runs
+// out. When cfg.resend is set and nothing has arrived after
+// cfg.resendAfter, the request is sent once more on a new stream, and the
+// budget is extended so the second request gets at least half a window.
+func awaitReply(w *inboxWatch, ackAt time.Time, cfg replyWait) (replyOutcome, error) {
+	type resendResult struct {
+		ackAt time.Time
+		err   error
+	}
+	var out replyOutcome
+	deadline := ackAt.Add(cfg.wait)
+	var resendCh chan resendResult // non-nil while the re-send is in flight
+	lastObserve := time.Time{}
+	for {
+		msg, err := w.poll()
+		if err != nil {
+			return out, err
+		}
+		if msg != nil {
+			out.reply = msg
+			out.waited = time.Since(ackAt)
+			return out, nil
+		}
+		now := time.Now()
+		if cfg.observe != nil && now.Sub(lastObserve) >= time.Second {
+			lastObserve = now
+			if n := cfg.observe(); n > out.replyConns {
+				out.replyConns = n
+			}
+		}
+		if resendCh != nil {
+			select {
+			case r := <-resendCh:
+				resendCh = nil
+				out.resendErr = r.err
+				if r.err == nil {
+					if extended := r.ackAt.Add(cfg.wait / 2); extended.After(deadline) {
+						deadline = extended
+					}
+				}
+			default:
+			}
+		}
+		if !out.resent && cfg.resend != nil && cfg.resendAfter > 0 && now.Sub(ackAt) >= cfg.resendAfter && now.Before(deadline) {
+			out.resent = true
+			resendCh = make(chan resendResult, 1)
+			go func(ch chan<- resendResult) {
+				at, err := cfg.resend()
+				ch <- resendResult{ackAt: at, err: err}
+			}(resendCh)
+		}
+		if now.After(deadline) && resendCh == nil {
+			out.waited = time.Since(ackAt)
+			return out, nil
+		}
+		time.Sleep(250 * time.Millisecond)
+	}
+}
+
+// replyTimeoutHint explains a request that was acknowledged but never
+// answered.
+func replyTimeoutHint(target string, out replyOutcome, firstContact bool) string {
+	var b strings.Builder
+	fmt.Fprintf(&b, "%s acknowledged the request but no reply arrived", target)
+	if out.resent {
+		if out.resendErr != nil {
+			fmt.Fprintf(&b, " (re-sending it failed: %v)", out.resendErr)
+		} else {
+			b.WriteString(" (the request was sent twice)")
+		}
+	}
+	b.WriteString(". ")
+	switch {
+	case out.replyConns > 0:
+		fmt.Fprintf(&b, "%s did dial its reply back (%d connection(s)) but no data arrived on it: the service gave up on the reply. ", target, out.replyConns)
+	case firstContact:
+		fmt.Fprintf(&b, "The reply comes back on a new connection from %s, and none was seen: the service is overloaded or dropped the reply while the new path was still settling. ", target)
+	default:
+		fmt.Fprintf(&b, "No reply connection from %s was seen. ", target)
+	}
+	b.WriteString("This is on the service side; retry in a minute, or use a longer --wait.")
+	return b.String()
+}
+
+// errNoReply is the error text scripts already match on.
+func errNoReply(target string, wait time.Duration) error {
+	return fmt.Errorf("no reply from %q within %s", target, wait)
+}
+
+// resendDelay is how long after the acknowledgement a first-contact
+// request with no reply is sent again: half the reply window, at most
+// 15 s, and never when the window is too short to fit a second request.
+func resendDelay(wait time.Duration) time.Duration {
+	d := wait / 2
+	if d > 15*time.Second {
+		d = 15 * time.Second
+	}
+	if d < 3*time.Second {
+		return 0
+	}
+	return d
+}
diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go
index 6b71e409..623c557e 100644
--- a/cmd/pilotctl/main.go
+++ b/cmd/pilotctl/main.go
@@ -650,6 +650,16 @@ func resolveHostnameToAddr(d *driver.Driver, hostname string) (protocol.Addr, ui
 //
 // Skip with --no-auto-handshake. Pre-#99 daemons don't have SubHandshakeWait;
 // we treat the IPC error as "wait unsupported" and proceed best-effort.
+//
+// With a daemon that has the reply window (feature "reply_window"), trust
+// is no longer a precondition for a reply: the daemon admits the dial-back
+// reply from a peer it just contacted. Branches 2 and 3 then no longer
+// block. Measured on clean runners, that block cost ~17-22 s of every
+// first query to list-agents (a direct handshake dial that times out, a
+// registry-relayed fallback, then the 5 s wait) before the data dial even
+// started. For a trusted agent the daemon still fires the handshake itself,
+// in parallel with the dial (DialConnection's inline auto-handshake). A
+// private peer we don't trust is still refused up front.
 func maybeAutoHandshake(d *driver.Driver, addr protocol.Addr, skip bool) {
 	if skip {
 		return
@@ -661,6 +671,7 @@ func maybeAutoHandshake(d *driver.Driver, addr protocol.Addr, skip bool) {
 			return
 		}
 	}
+	replyWindow := daemonHasFeature(d, "reply_window")
 
 	// Branch 2 — peer is in the embedded trusted-agents allowlist.
 	//
@@ -672,6 +683,9 @@ func maybeAutoHandshake(d *driver.Driver, addr protocol.Addr, skip bool) {
 	// at the inbound auto-accept path inside its NewService(), where the
 	// presented key IS available. See that repo, not this call site.
 	if name, ok := trustedagents.IsTrusted(addr.Node); ok {
+		if replyWindow {
+			return
+		}
 		if !jsonOutput {
 			fmt.Fprintf(os.Stderr, "establishing handshake with Trusted Agent %s (%s)...\n", name, addr)
 		}
@@ -729,6 +743,10 @@ func maybeAutoHandshake(d *driver.Driver, addr protocol.Addr, skip bool) {
 			fmt.Sprintf("run: pilotctl handshake %s", addr),
 			"refusing tunnel to private node %s without trust", addr)
 	}
+	if replyWindow {
+		// Public peer: it accepts our SYN, and our daemon admits its reply.
+		return
+	}
 	// Public peer — best-effort handshake so replies survive our local
 	// trust gate. Without this, a request/reply pattern (e.g. send-message
 	// --wait) gets its ACK back but the responder's reply SYN is silently
@@ -884,7 +902,11 @@ Flags:
   --type text|json|binary  payload encoding (default: text)
   --count <n>           send N times (default: 1)
   --reuse-conn          reuse the connection across --count sends (saves ~1 RTT)
-  --wait [<dur>]        reply wait only, after sending (default: 30s)
+  --wait [<dur>]        wait for a reply in the inbox (default timeout: 30s,
+                        counted from the receiver's acknowledgement)
+  --no-resend           on first contact, never send the request a second
+                        time (by default a request with no reply by mid-wait
+                        is re-sent once on a new stream)
   --timeout <dur>       total wall time, including connect, handshake and reply
   --trace               print per-step timing breakdown to stderr
   --no-auto-handshake   skip automatic trust handshake with known agents
@@ -4822,7 +4844,7 @@ func cmdSendMessage(args []string) {
 	flags, pos := parseFlags(args)
 	for name := range flags {
 		switch name {
-		case "data", "type", "count", "reuse-conn", "wait", "timeout", "trace", "no-auto-handshake", "enterprise-control", "governed-resource":
+		case "data", "type", "count", "reuse-conn", "wait", "timeout", "no-resend", "trace", "no-auto-handshake", "enterprise-control", "governed-resource":
 		default:
 			fatalCode("invalid_argument", "send-message: unknown flag --%s", name)
 		}
@@ -4842,7 +4864,7 @@ func cmdSendMessage(args []string) {
 		defer timer.Stop()
 	}
 	if len(pos) != 1 {
-		fatalCode("invalid_argument", "usage: pilotctl send-message <address|hostname> --data <text> [--type text|json|binary] [--trace] [--count <n>] [--reuse-conn] [--wait <dur>] [--enterprise-control <path> --governed-resource <receiver-resource>]")
+		fatalCode("invalid_argument", "usage: pilotctl send-message <address|hostname> --data <text> [--type text|json|binary] [--trace] [--count <n>] [--reuse-conn] [--wait <dur>] [--timeout <dur>] [--no-resend] [--enterprise-control <path> --governed-resource <receiver-resource>]")
 	}
 
 	sendCount := flagInt(flags, "count", 1)
@@ -4888,7 +4910,8 @@ func cmdSendMessage(args []string) {
 
 	d := connectDriver()
 	tracef("connectDriver")
-	defer d.Close()
+	// d may be replaced by a fresh connection when a dial is retried.
+	defer func() { d.Close() }()
 
 	target, err := parseAddrOrHostname(d, pos[0])
 	tracef("parseAddrOrHostname")
@@ -4902,6 +4925,14 @@ func cmdSendMessage(args []string) {
 	}
 	msgType := flagString(flags, "type", "text")
 
+	// First contact: the daemon holds no session with the peer yet, so this
+	// command builds the whole path (see firstcontact.go). Checked before
+	// anything is sent.
+	firstContact := false
+	if info, err := d.Info(); err == nil {
+		firstContact = !peerSessionUp(info, target.Node)
+	}
+
 	// Auto-handshake to peers in the embedded trusted-agents list.
 	// Best-effort: warns on stderr and continues if handshake fails.
 	maybeAutoHandshake(d, target, flagBool(flags, "no-auto-handshake"))
@@ -5030,18 +5061,50 @@ func cmdSendMessage(args []string) {
 	}
 	tracef("dial+send")
 
-	// Snapshot time before the send so --wait can find replies that arrive
-	// after this point even if the filesystem has 1-second mtime granularity.
-	inboxCutoff := time.Now().Add(-time.Second)
 	// agentHint is the resolved pilot address used to filter inbox replies
 	// against the "from" field written by the daemon when it saves the message.
 	agentHint := target.String()
+	// Snapshot the inbox before the send so --wait only takes a reply that
+	// arrives after it (never one already sitting there).
+	var watch *inboxWatch
+	if waitDur > 0 {
+		watch = newInboxWatch(agentHint)
+	}
 
 	if sendCount == 1 {
-		cl := dialOnce()
+		// On first contact a dial can fail while the path is still
+		// converging (a busy peer finishing the key exchange); dial once
+		// more before giving up, and say which step failed if it does.
+		dialStart := time.Now()
+		dialAttempts := 0
+		var cl *dataexchange.Client
+		for cl == nil {
+			dialAttempts++
+			c, err := dataexchange.Dial(d, target)
+			if err == nil {
+				cl = c
+				break
+			}
+			if firstContact && dialAttempts < 2 && isConvergingDialError(err) {
+				if isDriverSideTimeout(err) {
+					// The SDK gave up waiting, the daemon may still be
+					// dialing: use a fresh connection so a late reply to
+					// the first dial cannot be taken for the second.
+					d.Close()
+					d = connectDriver()
+				}
+				if !jsonOutput {
+					fmt.Fprintf(os.Stderr, "path to %s still converging (%v); dialing again...\n", pos[0], err)
+				}
+				continue
+			}
+			fatalHint("connection_failed", dialFailureHint(pos[0], err, dialAttempts, time.Since(dialStart)),
+				"cannot connect to %s (data exchange port %d)", target, protocol.PortDataExchange)
+		}
 		tracef("dataexchange.Dial")
 		defer cl.Close()
 		r := sendOne(cl, 0, false)
+		ackAt := time.Now()
 		if governedOutbound != nil {
 			if message, failed := r["error"].(string); failed {
 				fatalCode("permission_denied", "governed send-message: %s", message)
@@ -5058,31 +5121,82 @@ func cmdSendMessage(args []string) {
 		if len(traceEvents) > 0 {
 			result["trace"] = traceEvents
 		}
+		if firstContact {
+			result["first_contact"] = true
+		}
+		if dialAttempts > 1 {
+			result["dial_attempts"] = dialAttempts
+		}
 		if waitDur > 0 {
+			cfg := replyWait{wait: waitDur}
+			cfg.observe = func() int {
+				if info, err := d.Info(); err == nil {
+					return replyConnsFrom(info, agentHint)
+				}
+				return 0
+			}
+			// First contact: the service answers once, on a new connection
+			// it dials back, and that one reply is easily lost while its
+			// side of the new path is still settling (measured: the query
+			// then succeeded on the next attempt). Send the request once
+			// more on a new stream if nothing arrived by mid-window.
+			// Governed sends are never repeated (each needs its own
+			// authorization); --no-resend opts out for requests that are
+			// not safe to repeat.
+			_, sendFailed := r["error"]
+			if firstContact && !sendFailed && governedOutbound == nil && !flagBool(flags, "no-resend") {
+				if after := resendDelay(waitDur); after > 0 {
+					cfg.resendAfter = after
+					cfg.resend = func() (time.Time, error) {
+						rd, err := driver.Connect(getSocket())
+						if err != nil {
+							return time.Time{}, err
+						}
+						defer rd.Close()
+						c, err := dataexchange.Dial(rd, target)
+						if err != nil {
+							return time.Time{}, err
+						}
+						defer c.Close()
+						rr := sendOne(c, 1, false)
+						if e, failed := rr["error"].(string); failed {
+							return time.Time{}, errors.New(e)
+						}
+						if !jsonOutput {
+							fmt.Fprintf(os.Stderr, "no reply yet from %s; request sent again on a new stream\n", pos[0])
+						}
+						return time.Now(), nil
+					}
+				}
+			}
+			waitReply := func() map[string]interface{} {
+				stop := startWaitProgress("waiting for reply")
+				out, err := awaitReply(watch, ackAt, cfg)
+				stop()
+				if err != nil {
+					fatalCode("internal", "%v", err)
+				}
+				if out.resent {
+					result["resent"] = true
+				}
+				if out.reply == nil {
+					fatalHint("timeout", replyTimeoutHint(pos[0], out, firstContact), "%v", errNoReply(agentHint, waitDur))
+				}
+				result["reply_after_ms"] = out.waited.Milliseconds()
+				return out.reply
+			}
 			// Defer all output until the reply is in (or times out) so that
 			// --json emits a SINGLE document: a machine parser reading stdout
 			// must not see the send-result object followed by a second reply
 			// object. In JSON mode we fold the reply into one envelope; in
 			// human mode we still print the send result first, then the reply.
 			if jsonOutput {
-				stop := startWaitProgress("waiting for reply")
-				reply, err := waitForInboxReply(agentHint, inboxCutoff, waitDur)
-				stop()
-				if err != nil {
-					fatalCode("timeout", "%v", err)
-				}
-				result["reply"] = reply
+				result["reply"] = waitReply()
 				outputOK(result)
 			} else {
 				outputOK(result)
 				fmt.Fprintf(os.Stderr, "waiting for reply from %s (up to %s)...\n", pos[0], waitDur)
-				stop := startWaitProgress("waiting for reply")
-				reply, err := waitForInboxReply(agentHint, inboxCutoff, waitDur)
-				stop()
-				if err != nil {
-					fatalCode("timeout", "%v", err)
-				}
-				output(reply)
+				output(waitReply())
 			}
 		} else {
 			outputOK(result)
@@ -6919,52 +7033,25 @@ func cmdReceived(args []string) {
 	fmt.Println(sDim("filters: --since <dur> --limit <n> (0 = all) · clear: --clear [--before 24h] · json: --json"))
 }
 
-// cmdInbox lists or clears messages received via data exchange (port 1001).
 // waitForInboxReply polls ~/.pilot/inbox/ until a JSON file arrives that is
-// newer than cutoff and (if agentHint is non-empty) has a matching "agent"
-// field. Returns the parsed message or an error on timeout.
+// newer than cutoff and (if agentHint is non-empty) has a matching "from"
+// field, taking the oldest such file. Returns the parsed message or an
+// error on timeout. send-message --wait uses inboxWatch directly, which
+// also ignores files that were already in the inbox before the send.
 func waitForInboxReply(agentHint string, cutoff time.Time, timeout time.Duration) (map[string]interface{}, error) {
-	home, err := os.UserHomeDir()
-	if err != nil {
-		return nil, err
-	}
-	dir := filepath.Join(home, ".pilot", "inbox")
+	w := &inboxWatch{from: agentHint, cutoff: cutoff, seen: map[string]bool{}}
 	deadline := time.Now().Add(timeout)
 	for time.Now().Before(deadline) {
-		entries, err := os.ReadDir(dir)
-		if err != nil && !os.IsNotExist(err) {
-			return nil, fmt.Errorf("read inbox: %w", err)
+		msg, err := w.poll()
+		if err != nil {
+			return nil, err
 		}
-		for _, e := range entries {
-			if e.IsDir() {
-				continue
-			}
-			info, err := e.Info()
-			if err != nil {
-				continue
-			}
-			if !info.ModTime().After(cutoff) {
-				continue
-			}
-			data, err := os.ReadFile(filepath.Join(dir, e.Name()))
-			if err != nil {
-				continue
-			}
-			var msg map[string]interface{}
-			if json.Unmarshal(data, &msg) != nil {
-				continue
-			}
-			if agentHint != "" {
-				from, _ := msg["from"].(string)
-				if from != agentHint {
-					continue
-				}
-			}
+		if msg != nil {
 			return msg, nil
 		}
 		time.Sleep(250 * time.Millisecond)
 	}
-	return nil, fmt.Errorf("no reply from %q within %s", agentHint, timeout)
+	return nil, errNoReply(agentHint, timeout)
 }
 
 // inboxMessage is one parsed inbox entry plus its stable ID — the filename
diff --git a/cmd/pilotctl/zz_config_safety_test.go b/cmd/pilotctl/zz_config_safety_test.go
index 23d6ac38..064a4b69 100644
--- a/cmd/pilotctl/zz_config_safety_test.go
+++ b/cmd/pilotctl/zz_config_safety_test.go
@@ -120,7 +120,9 @@ func TestCmdSendMessageJSONWaitSingleDoc(t *testing.T) {
 	sd := newStreamDaemon(t)
 	home := sd.useDaemonNoRegistry(t)
 
-	// Seed an inbox reply whose "from" matches the resolved target address.
+	// The reply lands in the inbox after the request is sent, as a real
+	// one does: --wait ignores files that were already there before the
+	// send (an earlier request's reply, or a service's late duplicate).
 	inbox := filepath.Join(home, ".pilot", "inbox")
 	if err := os.MkdirAll(inbox, 0o700); err != nil {
 		t.Fatalf("mkdir inbox: %v", err)
@@ -131,14 +133,13 @@ func TestCmdSendMessageJSONWaitSingleDoc(t *testing.T) {
 	}
 	body, _ := json.Marshal(reply)
 	replyPath := filepath.Join(inbox, "TEXT-reply.json")
-	if err := os.WriteFile(replyPath, body, 0o600); err != nil {
-		t.Fatalf("write reply: %v", err)
-	}
-	// Ensure mtime is after the cutoff send-message computes (now-1s).
-	future := time.Now().Add(2 * time.Second)
-	if err := os.Chtimes(replyPath, future, future); err != nil {
-		t.Fatalf("chtimes: %v", err)
-	}
+	go func() {
+		deadline := time.Now().Add(5 * time.Second)
+		for sd.sendCount.Load() == 0 && time.Now().Before(deadline) {
+			time.Sleep(10 * time.Millisecond)
+		}
+		_ = os.WriteFile(replyPath, body, 0o600)
+	}()
 
 	out := captureStdout(t, func() {
 		withJSON(func() {
diff --git a/cmd/pilotctl/zz_firstcontact_test.go b/cmd/pilotctl/zz_firstcontact_test.go
new file mode 100644
index 00000000..52a9ffde
--- /dev/null
+++ b/cmd/pilotctl/zz_firstcontact_test.go
@@ -0,0 +1,302 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/driver"
+	"github.com/pilot-protocol/common/protocol"
+)
+
+func writeInboxMsg(t *testing.T, dir, name, from, data string) {
+	t.Helper()
+	body, _ := json.Marshal(map[string]any{"from": from, "data": data})
+	if err := os.WriteFile(filepath.Join(dir, name), body, 0o600); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func tempInbox(t *testing.T) string {
+	t.Helper()
+	tmp := withTempHomeFull(t)
+	dir := filepath.Join(tmp, ".pilot", "inbox")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	return dir
+}
+
+// A reply already in the inbox when the request is sent (an earlier
+// query's answer, or a service's late duplicate) is never taken for this
+// request's reply.
+func TestInboxWatchIgnoresFilesPresentBeforeTheSend(t *testing.T) {
+	dir := tempInbox(t)
+	writeInboxMsg(t, dir, "TEXT-20260924T100000.000-1.json", "0:0000.0002.BBE4", "stale")
+	w := newInboxWatch("0:0000.0002.BBE4")
+	if msg, err := w.poll(); err != nil || msg != nil {
+		t.Fatalf("poll before any new file = %v, %v; want nothing", msg, err)
+	}
+	writeInboxMsg(t, dir, "TEXT-20260924T100005.000-2.json", "0:0000.0002.BBE4", "fresh")
+	msg, err := w.poll()
+	if err != nil || msg == nil || msg["data"] != "fresh" {
+		t.Fatalf("poll = %v, %v; want the fresh reply", msg, err)
+	}
+}
+
+// With two new replies (a service that answers twice, or the answer to a
+// re-sent request), the oldest wins, whatever the directory order, and a
+// different sender's message is skipped.
+func TestInboxWatchTakesOldestReplyFromThePeer(t *testing.T) {
+	dir := tempInbox(t)
+	w := newInboxWatch("0:0000.0002.BBE4")
+	writeInboxMsg(t, dir, "TEXT-20260924T100007.000-3.json", "0:0000.0002.BBE4", "second")
+	writeInboxMsg(t, dir, "JSON-20260924T100006.000-2.json", "0:0000.0002.BBE4", "first")
+	writeInboxMsg(t, dir, "BINARY-20260924T100005.000-1.json", "0:0000.0009.0001", "other sender")
+	msg, err := w.poll()
+	if err != nil || msg == nil || msg["data"] != "first" {
+		t.Fatalf("poll = %v, %v; want the oldest reply from the peer", msg, err)
+	}
+}
+
+func TestAwaitReplyReturnsWithoutResendWhenReplyArrives(t *testing.T) {
+	dir := tempInbox(t)
+	w := newInboxWatch("peer")
+	var resends atomic.Int32
+	go func() {
+		time.Sleep(200 * time.Millisecond)
+		writeInboxMsg(t, dir, "TEXT-1-1.json", "peer", "ok")
+	}()
+	out, err := awaitReply(w, time.Now(), replyWait{
+		wait:        3 * time.Second,
+		resendAfter: time.Second,
+		resend:      func() (time.Time, error) { resends.Add(1); return time.Now(), nil },
+	})
+	if err != nil || out.reply == nil {
+		t.Fatalf("awaitReply = %+v, %v", out, err)
+	}
+	if out.resent || resends.Load() != 0 {
+		t.Fatal("request re-sent although the reply arrived in time")
+	}
+}
+
+// First contact: the first reply is lost; the request is sent once more
+// on a new stream at mid-window and its reply is taken.
+func TestAwaitReplyResendsOnceAndTakesTheSecondReply(t *testing.T) {
+	dir := tempInbox(t)
+	w := newInboxWatch("peer")
+	var resends atomic.Int32
+	start := time.Now()
+	out, err := awaitReply(w, start, replyWait{
+		wait:        4 * time.Second,
+		resendAfter: 600 * time.Millisecond,
+		resend: func() (time.Time, error) {
+			resends.Add(1)
+			go func() {
+				time.Sleep(300 * time.Millisecond)
+				writeInboxMsg(t, dir, "TEXT-2-1.json", "peer", "answer to the re-sent request")
+			}()
+			return time.Now(), nil
+		},
+	})
+	if err != nil || out.reply == nil {
+		t.Fatalf("awaitReply = %+v, %v", out, err)
+	}
+	if !out.resent || resends.Load() != 1 {
+		t.Fatalf("resent=%v resends=%d; want one re-send", out.resent, resends.Load())
+	}
+	if el := time.Since(start); el < 600*time.Millisecond || el > 3*time.Second {
+		t.Fatalf("reply taken after %s", el)
+	}
+}
+
+// A re-sent request gets at least half a window of its own, even when
+// sending it again took most of the original window.
+func TestAwaitReplyExtendsTheWindowForTheResentRequest(t *testing.T) {
+	tempInbox(t)
+	w := newInboxWatch("peer")
+	start := time.Now()
+	out, err := awaitReply(w, start, replyWait{
+		wait:        2 * time.Second,
+		resendAfter: 500 * time.Millisecond,
+		resend: func() (time.Time, error) {
+			time.Sleep(1200 * time.Millisecond) // a slow dial on a busy path
+			return time.Now(), nil
+		},
+	})
+	if err != nil || out.reply != nil || !out.resent {
+		t.Fatalf("awaitReply = %+v, %v", out, err)
+	}
+	// Re-send acknowledged at ~1.7 s: waits until ~2.7 s, not 2 s.
+	if el := time.Since(start); el < 2500*time.Millisecond {
+		t.Fatalf("gave up after %s, before the re-sent request had half a window", el)
+	}
+}
+
+func TestAwaitReplyRecordsResendFailure(t *testing.T) {
+	tempInbox(t)
+	w := newInboxWatch("peer")
+	out, err := awaitReply(w, time.Now(), replyWait{
+		wait:        time.Second,
+		resendAfter: 300 * time.Millisecond,
+		resend:      func() (time.Time, error) { return time.Time{}, errors.New("dial: dial timeout") },
+	})
+	if err != nil || out.reply != nil || !out.resent || out.resendErr == nil {
+		t.Fatalf("awaitReply = %+v, %v", out, err)
+	}
+	if h := replyTimeoutHint("list-agents", out, true); !strings.Contains(h, "re-sending it failed") {
+		t.Fatalf("hint does not mention the failed re-send: %s", h)
+	}
+}
+
+func TestAwaitReplyCountsReplyConnections(t *testing.T) {
+	tempInbox(t)
+	w := newInboxWatch("peer")
+	out, _ := awaitReply(w, time.Now(), replyWait{wait: 1200 * time.Millisecond, observe: func() int { return 2 }})
+	if out.replyConns != 2 {
+		t.Fatalf("replyConns = %d, want 2", out.replyConns)
+	}
+	if h := replyTimeoutHint("list-agents", out, true); !strings.Contains(h, "gave up on the reply") {
+		t.Fatalf("hint: %s", h)
+	}
+}
+
+func TestResendDelay(t *testing.T) {
+	for _, c := range []struct{ wait, want time.Duration }{
+		{30 * time.Second, 15 * time.Second},
+		{60 * time.Second, 15 * time.Second},
+		{10 * time.Second, 5 * time.Second},
+		{4 * time.Second, 0},
+	} {
+		if got := resendDelay(c.wait); got != c.want {
+			t.Errorf("resendDelay(%s) = %s, want %s", c.wait, got, c.want)
+		}
+	}
+}
+
+func TestReplyTimeoutHintNamesTheStep(t *testing.T) {
+	first := replyTimeoutHint("list-agents", replyOutcome{resent: true}, true)
+	for _, want := range []string{"acknowledged the request", "sent twice", "service side"} {
+		if !strings.Contains(first, want) {
+			t.Errorf("first-contact hint missing %q: %s", want, first)
+		}
+	}
+	if warm := replyTimeoutHint("list-agents", replyOutcome{}, false); !strings.Contains(warm, "No reply connection") {
+		t.Errorf("warm-path hint: %s", warm)
+	}
+}
+
+func TestDialFailureHint(t *testing.T) {
+	kx := dialFailureHint("list-agents", errors.New("dial: daemon: dial timeout: key exchange with peer did not complete"), 2, 55*time.Second)
+	if !strings.Contains(kx, "never completed the tunnel key exchange") || !strings.Contains(kx, "2 dial attempt") {
+		t.Errorf("key-exchange hint: %s", kx)
+	}
+	syn := dialFailureHint("list-agents", errors.New("dial: daemon: dial timeout"), 1, 17*time.Second)
+	if !strings.Contains(syn, "did not accept a connection on port 1001") {
+		t.Errorf("dial-timeout hint: %s", syn)
+	}
+}
+
+func TestDialErrorClassification(t *testing.T) {
+	daemonKX := errors.New("dial: daemon: dial timeout: key exchange with peer did not complete")
+	sdk := errors.New("dial: dial timeout")
+	refused := errors.New("dial: daemon: connection refused")
+	if !isConvergingDialError(daemonKX) || !isConvergingDialError(sdk) || isConvergingDialError(refused) {
+		t.Error("isConvergingDialError misclassifies")
+	}
+	if isDriverSideTimeout(daemonKX) || !isDriverSideTimeout(sdk) {
+		t.Error("isDriverSideTimeout misclassifies")
+	}
+}
+
+func TestInfoHelpers(t *testing.T) {
+	var info map[string]interface{}
+	raw := `{"features":["reply_window","dial_awaits_key"],
+	  "peer_list":[{"node_id":179172,"encrypted":true},{"node_id":7,"encrypted":false}],
+	  "conn_list":[{"local_port":1001,"remote_addr":"0:0000.0002.BBE4"},
+	               {"local_port":1001,"remote_addr":"0:0000.0000.0007"},
+	               {"local_port":49152,"remote_addr":"0:0000.0002.BBE4"}]}`
+	if err := json.Unmarshal([]byte(raw), &info); err != nil {
+		t.Fatal(err)
+	}
+	f := featuresOf(info)
+	if !f["reply_window"] || !f["dial_awaits_key"] || f["key_request"] {
+		t.Errorf("featuresOf = %v", f)
+	}
+	if !peerSessionUp(info, 179172) || peerSessionUp(info, 7) || peerSessionUp(info, 99) {
+		t.Error("peerSessionUp misreads the peer list")
+	}
+	if n := replyConnsFrom(info, "0:0000.0002.BBE4"); n != 1 {
+		t.Errorf("replyConnsFrom = %d, want 1", n)
+	}
+}
+
+// With a reply-window daemon, sending to a trusted agent (list-agents)
+// no longer blocks on a trust handshake first: pilotctl sends no
+// handshake request at all (the daemon fires one itself, alongside the
+// dial). Measured: that block cost ~17-22 s of the first query.
+func TestAutoHandshakeDoesNotBlockWithReplyWindow(t *testing.T) {
+	fd := newFakeDaemon(t)
+	fd.useDaemon(t)
+	daemonFeatureSet = nil
+	t.Cleanup(func() { daemonFeatureSet = nil })
+	fd.onJSON(tdCmdInfo, tdCmdInfoOK, `{"node_id":1,"features":["reply_window"]}`)
+	fd.onJSON(tdCmdHandshake, tdCmdHandshakeOK, `{"trusted":false}`)
+
+	d, err := driver.Connect(fd.path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	start := time.Now()
+	maybeAutoHandshake(d, protocol.Addr{Node: 179172}, false)
+	if el := time.Since(start); el > 2*time.Second {
+		t.Fatalf("maybeAutoHandshake blocked for %s", el)
+	}
+	fd.mu.Lock()
+	defer fd.mu.Unlock()
+	for _, f := range fd.received {
+		if f[0] == tdCmdHandshake && len(f) > 1 && f[1] == 0x01 {
+			t.Fatal("a handshake request was sent although the daemon admits the reply without trust")
+		}
+	}
+}
+
+// An older daemon (no features) keeps the old behaviour: the trusted
+// agent gets a handshake request first.
+func TestAutoHandshakeKeepsHandshakeForOlderDaemon(t *testing.T) {
+	fd := newFakeDaemon(t)
+	fd.useDaemon(t)
+	daemonFeatureSet = nil
+	t.Cleanup(func() { daemonFeatureSet = nil })
+	fd.onJSON(tdCmdInfo, tdCmdInfoOK, `{"node_id":1}`)
+	fd.onJSON(tdCmdHandshake, tdCmdHandshakeOK, `{"trusted":false}`)
+	prev := jsonOutput
+	jsonOutput = true
+	t.Cleanup(func() { jsonOutput = prev })
+
+	d, err := driver.Connect(fd.path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	maybeAutoHandshake(d, protocol.Addr{Node: 179172}, false)
+	fd.mu.Lock()
+	defer fd.mu.Unlock()
+	sent := false
+	for _, f := range fd.received {
+		if f[0] == tdCmdHandshake && len(f) > 1 && f[1] == 0x01 {
+			sent = true
+		}
+	}
+	if !sent {
+		t.Fatal("older daemon: the trusted-agent handshake request must still be sent")
+	}
+}
diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go
index 50dd2a9c..ffd4ef79 100644
--- a/pkg/daemon/daemon.go
+++ b/pkg/daemon/daemon.go
@@ -284,6 +284,16 @@ const (
 	MaxRetxAttempts      = 8                      // abandon connection after this many retransmissions
 	HeartbeatReregThresh = 3                      // heartbeat failures before re-registration
 	SYNBucketAge         = 10 * time.Second       // stale per-source SYN bucket reap threshold
+	// DialKeyExchangeWait bounds how long a dial to a peer we hold no
+	// session key for waits for the first-contact key exchange before it
+	// starts spending its SYN retry budget. Until the key is installed the
+	// SYN sits in the tunnel's pending queue (flushed the moment the key
+	// arrives), so retransmitting it only queued duplicates and burned
+	// retries: a key exchange that took 5-20 s against a busy service left
+	// the SYN phase too little time for the SYN-ACK. The wait plus the
+	// full retry budget (17.25 s) stays under the SDK's 30 s dial timeout
+	// for a normally reachable registry.
+	DialKeyExchangeWait = 10 * time.Second
 )
 
 // Zero-window probe constants.
@@ -407,6 +417,11 @@ type Daemon struct {
 	// IPC calls are NOT throttled.
 	autoHandshakeLastAttempt sync.Map
 
+	// outbound records the peers this node recently dialed or sent a
+	// trust handshake to, so the private-node SYN gate can admit their
+	// dial-back replies (see replywindow.go).
+	outbound outboundContacts
+
 	// network.* bus subscriber for daemon-internal reactions (managed
 	// engines + member-tag cache). Wired by subscribeNetworkInternalToBus
 	// during Start; torn down before tunnels in doStop. (T4.3.)
@@ -626,7 +641,16 @@ func New(cfg Config) *Daemon {
 	// machinery gives up on it (see onRekeyGaveUp / pathwatch.go).
 	d.tunnels.SetRekeyGaveUpHook(d.onRekeyGaveUp)
 	d.tunnels.SetTrustGate(d.admitDataPlanePeer)
-	d.tunnels.SetPeerTrustFn(d.isTrustedPeer)
+	// The key-exchange path asks this only to decide whether a
+	// tunnel.established event may name the peer. Local trust answers
+	// that without a registry round trip: isTrustedPeer's CheckTrust ran
+	// synchronously on the tunnel read loop for every new peer's PILA,
+	// delaying the key install hook (and the flush of the SYN queued
+	// behind it) by a registry RTT, seconds when the registry is shedding
+	// connections, and stalling every other inbound packet meanwhile. A
+	// peer trusted only through the registry now gets the redacted event,
+	// which is the conservative side.
+	d.tunnels.SetPeerTrustFn(d.handshakeTrusts)
 	d.ipc = NewIPCServer(cfg.SocketPath, d)
 	// HandshakeService is wired post-construction by the composition
 	// root via RegisterHandshakeService (T3.3 — handshake plugin moved
@@ -2244,6 +2268,8 @@ func (d *Daemon) HandshakeSendRequest(nodeID uint32, reason string) error {
 	if d.handshakes == nil {
 		return fmt.Errorf("handshake service not registered")
 	}
+	// The accept comes back as a dial to our port 444; open its window.
+	d.noteOutboundContact(nodeID)
 	if _, loaded := d.handshakeInFlight.LoadOrStore(nodeID, struct{}{}); loaded {
 		return ErrHandshakeInFlight
 	}
@@ -3247,6 +3273,12 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 		if !d.config.Public {
 			srcNode := pkt.Src.Node
 			trusted := d.handshakeTrusts(srcNode)
+			if !trusted && d.replyWindowAdmits(srcNode, pkt.DstPort) {
+				// A reply to a request we sent (replywindow.go). Admitted
+				// without the synchronous registry trust check.
+				trusted = true
+				slog.Debug("SYN admitted: reply from a peer we contacted", "src_node", srcNode, "dst_port", pkt.DstPort)
+			}
 			if !trusted && d.reg() != nil {
 				// Fall back to registry trust check (covers admin-set trust pairs + shared networks)
 				var err error
@@ -3820,6 +3852,9 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 	if err := d.ensureTunnel(dstAddr.Node); err != nil {
 		return nil, err
 	}
+	// Open the reply window before the first SYN: a fast service can dial
+	// its answer back before our dial even returns.
+	d.noteOutboundContact(dstAddr.Node)
 
 	// Compat mode: the daemon has no public UDP socket, so any direct
 	// SYN we send out can't reach the peer. Pre-flip the routing
@@ -3913,18 +3948,32 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 		Window:   conn.RecvWindow(),
 	}
 
+	// keyPending reports whether the tunnel still has no session key for
+	// the peer, i.e. every SYN we hand it is queued, not sent.
+	keyPending := func() bool {
+		return d.config.Encrypt && !d.tunnels.IsEncrypted(dstAddr.Node)
+	}
+	keyWaitUntil := time.Now().Add(DialKeyExchangeWait)
+	// synQueued is true while a copy of the SYN sits in the pending queue
+	// waiting for the key; retransmitting then would only queue duplicates
+	// that are all flushed at once when the key arrives.
+	synQueued := false
+
 	if err := d.tunnels.Send(dstAddr.Node, syn); err != nil {
-		// ErrPendingDropped means the SYN itself IS queued; only an older
-		// packet was dropped to make room. The tunnel is mid-handshake.
-		// Don't abort the dial — the SYN will go out as soon as the key
-		// exchange completes, and if not, the retry loop below will
-		// retransmit. Aborting here turns a transient handshake-startup
-		// condition into a hard "dial failed" for the user.
+		// ErrPendingDropped means the queue was full and this SYN was NOT
+		// queued. The tunnel is mid-handshake. Don't abort the dial — the
+		// retry loop below retransmits once the queue drains. Aborting
+		// here turns a transient handshake-startup condition into a hard
+		// "dial failed" for the user.
 		if !errors.Is(err, ErrPendingDropped) {
 			d.ports.RemoveConnection(conn.ID)
 			return nil, fmt.Errorf("send SYN: %w", err)
 		}
-		slog.Debug("dial: SYN queued during tunnel handshake (will retransmit)",
+		slog.Debug("dial: SYN not queued during tunnel handshake (will retransmit)",
+			"peer_node_id", dstAddr.Node, "dst_port", dstPort)
+	} else if keyPending() {
+		synQueued = true
+		slog.Debug("dial: SYN queued until the key exchange completes",
 			"peer_node_id", dstAddr.Node, "dst_port", dstPort)
 	}
 	conn.Mu.Lock()
@@ -4010,6 +4059,35 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 				return nil, protocol.ErrConnRefused
 			}
 		case <-timer.C:
+			if keyPending() {
+				// First contact: the key exchange is still running. Don't
+				// spend the retry budget (or queue duplicate SYNs) while
+				// no SYN can leave this node; the queued one goes out the
+				// moment the key is installed. Past DialKeyExchangeWait the
+				// dial counts retries as before, so it still ends.
+				if time.Now().Before(keyWaitUntil) {
+					if !synQueued {
+						synQueued = d.tunnels.Send(dstAddr.Node, syn) == nil
+					}
+					timer.Reset(rto)
+					continue
+				}
+			} else if !relayActive && retries < directRetries &&
+				d.config.BeaconAddr != "" && d.tunnels.KeyArrivedViaRelayOnly(dstAddr.Node) {
+				// The key came in over the relay and nothing has ever been
+				// decrypted from the peer's direct address: the relay is the
+				// only path it has proven. Start there instead of spending
+				// 1.75 s of direct retries on an address that never answered
+				// (e.g. a stale registry endpoint). relayProbeLoop keeps
+				// trying to upgrade the peer to direct in the background.
+				slog.Debug("dial: key arrived via relay only, dialing via relay",
+					"peer_node_id", dstAddr.Node, "dst_port", dstPort)
+				d.tunnels.SetRelayPeer(dstAddr.Node, true)
+				relayActive = true
+				relayActivatedHere = true
+				directRetries = 0
+				rto = DialInitialRTO
+			}
 			retries++
 
 			// Switch to relay mode after direct retries exhaust. Use
@@ -4026,6 +4104,7 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 			}
 
 			if retries > maxRetries {
+				keyMissing := keyPending()
 				d.ports.RemoveConnection(conn.ID)
 				// v1.9.1: a peer that didn't respond to the full retry
 				// budget (direct + relay) is reachably dead. Invalidate
@@ -4046,13 +4125,23 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 				// Feeds the rx-watchdog's partial-wedge detector: a run of
 				// these to distinct peers means our outbound is wedged.
 				d.consecutiveDialTimeouts.Add(1)
+				if keyMissing {
+					// Not one SYN left this node: the peer never completed
+					// the key exchange. Say so, so callers can tell "busy
+					// or unreachable peer" from "peer refused the port".
+					return nil, ErrDialKeyExchange
+				}
 				return nil, protocol.ErrDialTimeout
 			}
-			// Resend SYN (uses relay if relayActive)
+			// Resend SYN (uses relay if relayActive). While the key is
+			// still missing, one queued copy is enough.
 			conn.Mu.Lock()
 			syn.Seq = conn.SendSeq - 1
 			conn.Mu.Unlock()
-			d.tunnels.Send(dstAddr.Node, syn)
+			if !keyPending() || !synQueued {
+				err := d.tunnels.Send(dstAddr.Node, syn)
+				synQueued = err == nil && keyPending()
+			}
 			rto = rto * 2 // exponential backoff
 			if rto > DialMaxRTO {
 				rto = DialMaxRTO
@@ -4062,6 +4151,12 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 	}
 }
 
+// ErrDialKeyExchange is returned by a dial whose peer never completed the
+// tunnel key exchange, so no SYN could be sent at all. It wraps
+// protocol.ErrDialTimeout: callers that only check for a dial timeout
+// keep working, and the message names the actual cause.
+var ErrDialKeyExchange = fmt.Errorf("%w: key exchange with peer did not complete", protocol.ErrDialTimeout)
+
 // NagleTimeout is the maximum time to buffer small writes before flushing.
 const NagleTimeout = 40 * time.Millisecond
 
@@ -5621,6 +5716,9 @@ func (d *Daemon) idleSweepLoop() {
 			// Reap stale per-source SYN rate limit buckets
 			d.reapPerSrcSYN()
 
+			// Forget reply windows that have expired.
+			d.outbound.prune(time.Now())
+
 			// Reap stale "queue full" log-throttle entries so
 			// lastPendDropLog doesn't accumulate one slot per
 			// unique-peer-ever-throttled.
diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go
index 14b8a3c8..9256af56 100644
--- a/pkg/daemon/ipc.go
+++ b/pkg/daemon/ipc.go
@@ -1108,6 +1108,20 @@ func (s *IPCServer) handleBroadcast(conn *ipcConn, reqID uint64, payload []byte)
 	}
 }
 
+// daemonFeatures lists behaviours a client can rely on from this daemon,
+// reported in the info reply so a newer pilotctl can adapt to an older
+// daemon (and the reverse) without comparing version strings:
+//
+//   - reply_window: a private node admits a dial-back reply (SYN to port
+//     1001 or 444) from a peer it contacted in the last few minutes, so
+//     request/reply no longer needs a trust handshake first.
+//   - dial_awaits_key: a dial to a peer with no session key yet waits for
+//     the key exchange before spending its SYN retries, and reports
+//     "key exchange with peer did not complete" when it never finishes.
+//   - key_request: an unanswered first-contact key exchange also sends a
+//     key request that every released daemon answers.
+var daemonFeatures = []string{"reply_window", "dial_awaits_key", "key_request"}
+
 func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) {
 	info := s.daemon.Info()
 
@@ -1186,6 +1200,7 @@ func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) {
 		"relay_peer_count":          info.RelayPeerCount,
 		"beacon_addr":               info.BeaconAddr,
 		"motd":                      info.MOTD,
+		"features":                  daemonFeatures,
 		"transport":                 info.Transport,
 	})
 	if err != nil {
diff --git a/pkg/daemon/keyexchange/bootstrap.go b/pkg/daemon/keyexchange/bootstrap.go
index 59ac28a0..1f7d5af3 100644
--- a/pkg/daemon/keyexchange/bootstrap.go
+++ b/pkg/daemon/keyexchange/bootstrap.go
@@ -10,11 +10,21 @@ import (
 // SendKeyExchangeToNode sends an authenticated key exchange if identity
 // is available, otherwise falls back to unauthenticated.
 //
+// When we hold no session key for the peer and either an earlier send is
+// still unanswered (KeyRequestAfterSends) or MarkKeyRequestDue flagged
+// the peer, the authenticated frame is followed by an unauthenticated
+// one (PILK) acting as a key request: a peer that already knows our
+// identity rejects it and answers with its own PILA, retransmitted until
+// ours arrives. This breaks the first-contact deadlock where the peer's
+// single reply was lost and it treats our same-key retransmits as
+// keepalives. keyRequest reports whether that PILK was sent, so the
+// tunnel can give it the same relay copy as the PILA.
+//
 // This function carries the single annotated bootstrap-exception site
 // (see the marker comment inside the body). After Stage 2 sub-pass 2,
 // the canonical home for the marker is here in keyexchange/bootstrap.go;
 // layers.yaml's bootstrap_exception.allowed_call_sites tracks this path.
-func (m *Manager) SendKeyExchangeToNode(peerNodeID uint32) {
+func (m *Manager) SendKeyExchangeToNode(peerNodeID uint32) (keyRequest bool) {
 	// BOOTSTRAP-EXCEPTION: bypasses L6 envelope
 	// The peer key is not yet established, so the L6 AEAD wrap is
 	// impossible here. This is the single annotated bootstrap site
@@ -23,7 +33,7 @@ func (m *Manager) SendKeyExchangeToNode(peerNodeID uint32) {
 	// marker elsewhere — the checker fails if more than one site
 	// carries it. See docs/architecture/05-VERIFICATION.md §3 P8.
 	if m.sender == nil {
-		return
+		return false
 	}
 
 	var addr *net.UDPAddr
@@ -32,25 +42,40 @@ func (m *Manager) SendKeyExchangeToNode(peerNodeID uint32) {
 	}
 
 	hasIdentity := m.HasIdentity()
-	frame := m.BuildUnauthFrame()
-	if frame == nil {
-		return
+	unauthFrame := m.BuildUnauthFrame()
+	if unauthFrame == nil {
+		return false
 	}
-
+	frame := unauthFrame
+	authed := false
 	if hasIdentity {
-		authFrame := m.BuildAuthFrame()
-		if authFrame != nil {
+		if authFrame := m.BuildAuthFrame(); authFrame != nil {
 			frame = authFrame
+			authed = true
 		}
 	}
+	// Decide before MarkPendingRekey bumps the attempt count: the key
+	// request rides on retransmits, never on the first send.
+	keyRequest = authed && !m.env.Has(peerNodeID) && m.takeKeyRequest(peerNodeID)
 
 	if err := m.sender(peerNodeID, addr, frame); err != nil {
 		slog.Error("send key exchange failed", "peer_node_id", peerNodeID, "error", err)
-		return
+		return false
+	}
+	if keyRequest {
+		if err := m.sender(peerNodeID, addr, unauthFrame); err != nil {
+			slog.Debug("key request send failed", "peer_node_id", peerNodeID, "error", err)
+			keyRequest = false
+		} else {
+			m.keyRequestsSent.Add(1)
+			slog.Debug("key request sent (no session key, key exchange unanswered)",
+				"peer_node_id", peerNodeID)
+		}
 	}
 
 	// P1-010 tunnel-state half: register that we're waiting on a reply,
 	// so rekeyRetransmitLoop can retransmit if the peer's response (or
 	// our request) was dropped under loss.
 	m.MarkPendingRekey(peerNodeID)
+	return keyRequest
 }
diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go
index 144a4837..729ed3c9 100644
--- a/pkg/daemon/keyexchange/keyexchange.go
+++ b/pkg/daemon/keyexchange/keyexchange.go
@@ -147,6 +147,25 @@ const (
 
 	NegativePubKeyCacheTTL = 3 * time.Second
 	MaxNegPubKeyEntries    = 4096
+
+	// KeyRequestAfterSends is how many of our key-exchange frames to a peer
+	// may go unanswered, while we hold no session key for it, before each
+	// further send also carries a key request: an unauthenticated PILK
+	// frame next to the PILA. See SendKeyExchangeToNode.
+	//
+	// Why: a node answers a new peer's PILA exactly once and clears its
+	// own retransmit state (HandleAuthFrame, the !hadCrypto branch). If
+	// that one reply is lost, every retransmit of ours carries the same
+	// X25519 key, so the peer treats it as a same-session keepalive and
+	// stays silent: the peer holds a session for us, we hold none, and
+	// nothing on either side recovers it until the peer's path watchdog
+	// drops its half (which our own PILAs keep postponing). Every
+	// released daemon answers a PILK from a peer whose identity it knows
+	// by rejecting it and sending its authenticated PILA instead
+	// (HandleUnauthFrame), and that reply is marked pending, so the peer
+	// retransmits it until our PILA arrives. The PILK installs nothing on
+	// such a peer and costs one 40-byte frame per retransmit.
+	KeyRequestAfterSends = 1
 )
 
 // PendingRekeyState tracks a key-exchange we sent and are waiting on.
@@ -248,6 +267,13 @@ type Manager struct {
 	// KeyExchangeReplyMinInterval to break the symmetric ping-pong
 	// described on that constant.
 	lastKeyExchangeReply map[uint32]time.Time
+	// keyRequestDue marks peers whose next SendKeyExchangeToNode must
+	// carry a key request (PILK) regardless of the attempt count: set by
+	// MarkKeyRequestDue when the peer is known to hold a session we lack
+	// (it sent us an encrypted frame we have no key for).
+	keyRequestDue map[uint32]struct{}
+	// keyRequestsSent counts key requests (PILK nudges) sent. Metric only.
+	keyRequestsSent atomic.Uint64
 
 	// PILOT-344: per-peer whitelist for the reply-interval gate.
 	// Whitelisted peers always pass MarkReplyKeyExchangeSent so trusted
@@ -285,6 +311,7 @@ func New(store *Store) *Manager {
 		lastInboundDecrypt:   make(map[uint32]time.Time),
 		rekeyGaveUp:          make(map[uint32]time.Time),
 		lastKeyExchangeReply: make(map[uint32]time.Time),
+		keyRequestDue:        make(map[uint32]struct{}),
 	}
 }
 
@@ -718,5 +745,38 @@ func (m *Manager) RemovePeer(nodeID uint32) {
 	delete(m.pendingRekey, nodeID)
 	delete(m.lastInboundDecrypt, nodeID)
 	delete(m.lastKeyExchangeReply, nodeID)
+	delete(m.keyRequestDue, nodeID)
+	m.rkPendingMu.Unlock()
+}
+
+// MarkKeyRequestDue makes the next SendKeyExchangeToNode to peerNodeID
+// also send a key request (PILK), provided we still hold no session key
+// for it then. The tunnel calls it when the peer sends us an encrypted
+// frame we cannot decrypt: proof that the peer holds a session for us
+// that we lack, which a same-key PILA alone cannot repair (see
+// KeyRequestAfterSends).
+func (m *Manager) MarkKeyRequestDue(peerNodeID uint32) {
+	m.rkPendingMu.Lock()
+	if len(m.keyRequestDue) < MaxNegPubKeyEntries {
+		m.keyRequestDue[peerNodeID] = struct{}{}
+	}
 	m.rkPendingMu.Unlock()
 }
+
+// takeKeyRequest reports whether the send about to go out to peerNodeID
+// should carry a key request, and consumes a MarkKeyRequestDue mark.
+// Caller has already checked that no session key is installed.
+func (m *Manager) takeKeyRequest(peerNodeID uint32) bool {
+	m.rkPendingMu.Lock()
+	defer m.rkPendingMu.Unlock()
+	if _, due := m.keyRequestDue[peerNodeID]; due {
+		delete(m.keyRequestDue, peerNodeID)
+		return true
+	}
+	st, ok := m.pendingRekey[peerNodeID]
+	return ok && st.Attempts >= KeyRequestAfterSends
+}
+
+// KeyRequestsSent returns how many key requests (PILK nudges) this
+// manager has sent.
+func (m *Manager) KeyRequestsSent() uint64 { return m.keyRequestsSent.Load() }
diff --git a/pkg/daemon/keyexchange/zz_key_request_test.go b/pkg/daemon/keyexchange/zz_key_request_test.go
new file mode 100644
index 00000000..4008412a
--- /dev/null
+++ b/pkg/daemon/keyexchange/zz_key_request_test.go
@@ -0,0 +1,162 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package keyexchange_test
+
+// Key request (KeyRequestAfterSends): when our key exchange to a peer we
+// hold no session key for goes unanswered, each retransmit also sends a
+// PILK, which every released daemon answers with its authenticated PILA
+// (retransmitted until ours arrives). This breaks the first-contact
+// deadlock where the peer's single reply was lost and it treats our
+// same-key PILAs as keepalives.
+
+import (
+	"bytes"
+	"net"
+	"sync"
+	"testing"
+
+	"github.com/pilot-protocol/common/protocol"
+)
+
+type magicRecorder struct {
+	mu     sync.Mutex
+	frames [][]byte
+}
+
+func (r *magicRecorder) send(_ uint32, _ *net.UDPAddr, frame []byte) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.frames = append(r.frames, append([]byte(nil), frame...))
+	return nil
+}
+
+func (r *magicRecorder) take() (pila, pilk int, all [][]byte) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, f := range r.frames {
+		switch {
+		case bytes.Equal(f[:4], protocol.TunnelMagicAuthEx[:]):
+			pila++
+		case bytes.Equal(f[:4], protocol.TunnelMagicKeyEx[:]):
+			pilk++
+		}
+	}
+	all = r.frames
+	r.frames = nil
+	return pila, pilk, all
+}
+
+func TestKeyRequestRidesOnRetransmitsOnly(t *testing.T) {
+	t.Parallel()
+	a := newPeer(t, 1)
+	rec := &magicRecorder{}
+	a.mgr.SetSender(rec.send)
+
+	if a.mgr.SendKeyExchangeToNode(2) {
+		t.Fatal("first send reported a key request")
+	}
+	if pila, pilk, _ := rec.take(); pila != 1 || pilk != 0 {
+		t.Fatalf("first send: %d PILA, %d PILK; want 1, 0", pila, pilk)
+	}
+
+	// Unanswered: the retransmit carries the key request, PILA first.
+	if !a.mgr.SendKeyExchangeToNode(2) {
+		t.Fatal("retransmit did not report a key request")
+	}
+	pila, pilk, frames := rec.take()
+	if pila != 1 || pilk != 1 {
+		t.Fatalf("retransmit: %d PILA, %d PILK; want 1, 1", pila, pilk)
+	}
+	if !bytes.Equal(frames[0][:4], protocol.TunnelMagicAuthEx[:]) {
+		t.Fatal("the PILA must go out before the key request")
+	}
+	if got := a.mgr.KeyRequestsSent(); got != 1 {
+		t.Fatalf("KeyRequestsSent = %d, want 1", got)
+	}
+}
+
+func TestKeyRequestDueSendsOnFirstTry(t *testing.T) {
+	t.Parallel()
+	a := newPeer(t, 1)
+	rec := &magicRecorder{}
+	a.mgr.SetSender(rec.send)
+
+	a.mgr.MarkKeyRequestDue(3)
+	if !a.mgr.SendKeyExchangeToNode(3) {
+		t.Fatal("a due key request was not sent")
+	}
+	if pila, pilk, _ := rec.take(); pila != 1 || pilk != 1 {
+		t.Fatalf("%d PILA, %d PILK; want 1, 1", pila, pilk)
+	}
+	// The mark is consumed; the pending count alone decides from now on
+	// (and a retransmit is due, so the next send still carries one).
+	a.mgr.ClearPendingRekey(3)
+	if a.mgr.SendKeyExchangeToNode(3) {
+		t.Fatal("the due mark was not consumed")
+	}
+}
+
+func TestNoKeyRequestOnceSessionInstalled(t *testing.T) {
+	t.Parallel()
+	a := newPeer(t, 1)
+	b := newPeer(t, 2)
+	crossWireVerifyFuncs(a, b)
+	rec := &magicRecorder{}
+	a.mgr.SetSender(rec.send)
+	b.mgr.SetSender(func(uint32, *net.UDPAddr, []byte) error { return nil })
+
+	// A has B's key: B's PILA arrived.
+	from := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 4000}
+	if !a.mgr.HandleAuthFrame(b.mgr.BuildAuthFrame()[4:], from, false) {
+		t.Fatal("B's PILA rejected")
+	}
+	rec.take()
+	a.mgr.MarkKeyRequestDue(2)
+	a.mgr.SendKeyExchangeToNode(2)
+	a.mgr.SendKeyExchangeToNode(2)
+	if _, pilk, _ := rec.take(); pilk != 0 {
+		t.Fatalf("%d key requests sent to a peer we hold a key for; want 0", pilk)
+	}
+}
+
+// TestPeerAnswersKeyRequestWithRetransmittedPILA pins the peer-side
+// behaviour the key request relies on (it has been the same in every
+// released daemon): a PILK from a peer whose identity we know is
+// rejected and answered with our PILA, and that answer is marked pending
+// so it is retransmitted until the requester's PILA arrives.
+func TestPeerAnswersKeyRequestWithRetransmittedPILA(t *testing.T) {
+	t.Parallel()
+	client := newPeer(t, 10)
+	service := newPeer(t, 20)
+	crossWireVerifyFuncs(client, service)
+	rec := &magicRecorder{}
+	service.mgr.SetSender(rec.send)
+	client.mgr.SetSender(func(uint32, *net.UDPAddr, []byte) error { return nil })
+	from := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 4000}
+
+	// First contact: the service installs our key and answers once, then
+	// clears its own retransmit state. That answer is the one lost.
+	if !service.mgr.HandleAuthFrame(client.mgr.BuildAuthFrame()[4:], from, false) {
+		t.Fatal("client PILA rejected")
+	}
+	rec.take()
+	if service.mgr.PendingRekeyHas(10) {
+		t.Fatal("setup: the first answer is not retransmitted")
+	}
+
+	// The key request.
+	if service.mgr.HandleUnauthFrame(client.mgr.BuildUnauthFrame()[4:], from, false) {
+		t.Fatal("a PILK from a peer with a known identity must not install anything")
+	}
+	if pila, _, _ := rec.take(); pila != 1 {
+		t.Fatalf("service answered the key request with %d PILA, want 1", pila)
+	}
+	if !service.mgr.PendingRekeyHas(10) {
+		t.Fatal("the answer to a key request must be retransmitted until the requester's PILA arrives")
+	}
+	// The requester's PILA (sent when it installs the service's key) ends it.
+	service.mgr.HandleAuthFrame(client.mgr.BuildAuthFrame()[4:], from, false)
+	if service.mgr.PendingRekeyHas(10) {
+		t.Fatal("the requester's PILA must clear the service's retransmits")
+	}
+}
diff --git a/pkg/daemon/keyexchange/zz_keyexchange_test.go b/pkg/daemon/keyexchange/zz_keyexchange_test.go
index 1b590935..c04ef6a8 100644
--- a/pkg/daemon/keyexchange/zz_keyexchange_test.go
+++ b/pkg/daemon/keyexchange/zz_keyexchange_test.go
@@ -512,7 +512,7 @@ func TestRekeyRetransmitRetransmits(t *testing.T) {
 	t.Parallel()
 
 	p := newPeer(t, 7)
-	var sends atomic.Int32
+	var sends, keyRequests atomic.Int32
 	p.mgr.SetSender(func(peer uint32, _ *net.UDPAddr, frame []byte) error {
 		if peer != 50 {
 			t.Errorf("retransmit dest: got %d want 50", peer)
@@ -520,6 +520,13 @@ func TestRekeyRetransmitRetransmits(t *testing.T) {
 		if len(frame) == 0 {
 			t.Errorf("retransmit frame is empty")
 		}
+		// No session key is installed for peer 50, so the retransmit
+		// also carries a key request (PILK) next to the PILA; count the
+		// PILA retransmits on their own.
+		if bytes.Equal(frame[0:4], protocol.TunnelMagicKeyEx[:]) {
+			keyRequests.Add(1)
+			return nil
+		}
 		sends.Add(1)
 		return nil
 	})
@@ -536,6 +543,9 @@ func TestRekeyRetransmitRetransmits(t *testing.T) {
 	if got := sends.Load(); got != 1 {
 		t.Fatalf("retransmit send count: got %d want 1", got)
 	}
+	if got := keyRequests.Load(); got != 1 {
+		t.Fatalf("key requests with the retransmit: got %d want 1", got)
+	}
 	if attempts := p.mgr.PendingRekeyAttempts(50); attempts != 3 {
 		t.Fatalf("after retransmit Attempts=%d want 3", attempts)
 	}
diff --git a/pkg/daemon/replywindow.go b/pkg/daemon/replywindow.go
new file mode 100644
index 00000000..718152b8
--- /dev/null
+++ b/pkg/daemon/replywindow.go
@@ -0,0 +1,107 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+import (
+	"sync"
+	"time"
+
+	"github.com/pilot-protocol/common/protocol"
+)
+
+// Reply window: the private-node SYN trust gate admits a service's reply.
+//
+// A request/reply service answers on a NEW connection it dials back to
+// the requester: data exchange replies arrive as a SYN to our port 1001,
+// and a trust-handshake accept as a SYN to our port 444. A private node
+// (the default) drops a SYN from any peer it does not trust yet, and
+// with no RST, so on first contact the reply to our own request was
+// dropped until the trust handshake with that peer happened to finish.
+// Measured on clean GitHub runners: about 50 "SYN rejected: untrusted
+// source" per failed first query to list-agents, and the reply was
+// lost for good because the service does not retry for long.
+//
+// The gate now behaves like a stateful firewall: for replyWindow after
+// this node dials a peer (or sends it a trust handshake), a SYN from that
+// peer to one of the reply ports is admitted. Only the peer we contacted,
+// only those two ports, and only for a bounded time; everything else
+// still needs trust. The admitted connection is an ordinary inbound
+// connection to a service that is already listening, so it reaches
+// nothing a trusted peer could not.
+
+// replyWindow is how long after we contact a peer its SYNs to the reply
+// ports pass the private-node trust gate.
+const replyWindow = 5 * time.Minute
+
+// maxReplyWindowPeers bounds the reply-window table. It is filled only
+// by our own outbound dials, never by inbound traffic, so the bound is a
+// backstop: once full (after pruning expired entries) new peers simply
+// fall back to the regular trust check.
+const maxReplyWindowPeers = 4096
+
+// replyWindowPort reports whether an inbound SYN to port is a reply a
+// service may send back on a new connection.
+func replyWindowPort(port uint16) bool {
+	return port == protocol.PortDataExchange || port == protocol.PortHandshake
+}
+
+// outboundContacts records when this node last initiated contact with
+// each peer (nodeID -> time of the last outbound dial or handshake).
+type outboundContacts struct {
+	mu sync.Mutex
+	at map[uint32]time.Time
+}
+
+func (o *outboundContacts) note(peer uint32, now time.Time) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.at == nil {
+		o.at = make(map[uint32]time.Time)
+	}
+	if _, ok := o.at[peer]; !ok && len(o.at) >= maxReplyWindowPeers {
+		o.pruneLocked(now)
+		if len(o.at) >= maxReplyWindowPeers {
+			return
+		}
+	}
+	o.at[peer] = now
+}
+
+func (o *outboundContacts) recent(peer uint32, now time.Time) bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	t, ok := o.at[peer]
+	return ok && now.Sub(t) < replyWindow
+}
+
+func (o *outboundContacts) prune(now time.Time) {
+	o.mu.Lock()
+	o.pruneLocked(now)
+	o.mu.Unlock()
+}
+
+func (o *outboundContacts) pruneLocked(now time.Time) {
+	for peer, t := range o.at {
+		if now.Sub(t) >= replyWindow {
+			delete(o.at, peer)
+		}
+	}
+}
+
+func (o *outboundContacts) len() int {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return len(o.at)
+}
+
+// noteOutboundContact records that we just initiated contact with peer
+// (a dial, or a trust handshake), opening its reply window.
+func (d *Daemon) noteOutboundContact(peer uint32) {
+	d.outbound.note(peer, time.Now())
+}
+
+// replyWindowAdmits reports whether a SYN from srcNode to dstPort is a
+// reply to contact we initiated within replyWindow.
+func (d *Daemon) replyWindowAdmits(srcNode uint32, dstPort uint16) bool {
+	return replyWindowPort(dstPort) && d.outbound.recent(srcNode, time.Now())
+}
diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go
index d23fdd9d..ea9d1961 100644
--- a/pkg/daemon/tunnel.go
+++ b/pkg/daemon/tunnel.go
@@ -204,6 +204,13 @@ type TunnelManager struct {
 	// for key exchange. Exposed so operators can tell a congested overlay
 	// apart from a silent crypto stall.
 	PendingDrops uint64
+
+	// keyViaRelay records, per peer, whether the installed session key
+	// arrived in a relay-delivered key exchange (nodeID -> bool). With no
+	// direct decrypt since (routing.LastDirectRecv), the relay is the
+	// only path this peer has proven, so a dial starts there instead of
+	// spending its direct retries on an address that never answered.
+	keyViaRelay sync.Map
 }
 
 type IncomingPacket struct {
@@ -628,6 +635,12 @@ func (tm *TunnelManager) maybeRequestRekey(peerNodeID uint32, from *net.UDPAddr)
 		}
 	}
 
+	// The peer is sending us encrypted frames, so it holds a session for
+	// us. If we hold none, a same-key PILA alone is answered as a
+	// keepalive and never repairs that; ask for the peer's key as well.
+	if !tm.envelope.Has(peerNodeID) {
+		tm.kx.MarkKeyRequestDue(peerNodeID)
+	}
 	tm.sendKeyExchangeToNode(peerNodeID)
 	return true
 }
@@ -1460,6 +1473,9 @@ func (tm *TunnelManager) onKeyInstalled(ev keyexchange.PostInstallEvent) {
 	// and the per-peer 1s reply cooldown holds the ping-pong back but
 	// never lets the staleness flag clear.
 	tm.recordInboundDecrypt(peerNodeID)
+	if !ev.HadCrypto || ev.KeyChanged {
+		tm.keyViaRelay.Store(peerNodeID, fromRelay)
+	}
 
 	tm.mu.Lock()
 	if !fromRelay {
@@ -1756,7 +1772,7 @@ func (tm *TunnelManager) deriveSecret(peerPubKeyBytes []byte) (*peerCrypto, erro
 // keyexchange.Manager.SendKeyExchangeToNode (which carries the
 // BOOTSTRAP-EXCEPTION marker — moved with the function).
 func (tm *TunnelManager) sendKeyExchangeToNode(peerNodeID uint32) {
-	tm.kx.SendKeyExchangeToNode(peerNodeID)
+	keyRequest := tm.kx.SendKeyExchangeToNode(peerNodeID)
 
 	// Dual-NAT convergence fix: when a peer is not (yet) relay-flagged but
 	// a beacon is available, ALSO push the key-exchange via relay. For two
@@ -1784,6 +1800,15 @@ func (tm *TunnelManager) sendKeyExchangeToNode(peerNodeID uint32) {
 	if err := tm.routing.SendRelayFrame(peerNodeID, frame); err != nil {
 		slog.Debug("kx relay-copy send failed", "peer_node_id", peerNodeID, "error", err)
 	}
+	// The key request (see keyexchange.KeyRequestAfterSends) gets the same
+	// relay copy: the peer answers on whichever path it recorded for us.
+	if keyRequest {
+		if req := tm.kx.BuildUnauthFrame(); req != nil {
+			if err := tm.routing.SendRelayFrame(peerNodeID, req); err != nil {
+				slog.Debug("key request relay-copy send failed", "peer_node_id", peerNodeID, "error", err)
+			}
+		}
+	}
 }
 
 // markPendingRekey is the legacy shim for keyexchange.Manager.MarkPendingRekey.
@@ -2185,6 +2210,20 @@ func (tm *TunnelManager) RemovePeer(nodeID uint32) {
 
 	// L5-owned per-peer state (peerPubKeys, pendingRekey, lastInboundDecrypt).
 	tm.kx.RemovePeer(nodeID)
+	tm.keyViaRelay.Delete(nodeID)
+}
+
+// KeyArrivedViaRelayOnly reports whether the peer's session key was
+// installed from a relay-delivered key exchange and nothing has been
+// decrypted from the peer's direct path since: the relay is the only
+// path the peer has proven to work.
+func (tm *TunnelManager) KeyArrivedViaRelayOnly(nodeID uint32) bool {
+	v, ok := tm.keyViaRelay.Load(nodeID)
+	if !ok {
+		return false
+	}
+	viaRelay, _ := v.(bool)
+	return viaRelay && tm.routing.LastDirectRecv(nodeID).IsZero()
 }
 
 // HasPeer checks if we have a tunnel to a node.
diff --git a/pkg/daemon/zz_first_contact_test.go b/pkg/daemon/zz_first_contact_test.go
new file mode 100644
index 00000000..073866fa
--- /dev/null
+++ b/pkg/daemon/zz_first_contact_test.go
@@ -0,0 +1,519 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+// First contact from a fresh node to a service (the onboarding query to
+// list-agents). Measured on clean GitHub runners, the first query failed
+// most of the time; three daemon-side causes are pinned here:
+//
+//  1. Lost key-exchange reply deadlock. A node answers a new peer's PILA
+//     once. If that reply is lost, our retransmits carry the same X25519
+//     key, the peer treats them as same-session keepalives and never
+//     answers again: it holds a session for us, we hold none. Reproduced
+//     against the live v1.13.5 fleet (list-agents, randomfox): with the
+//     first reply dropped, no key in ~100 s, 4 of 4 runs; with the key
+//     request, the key arrived 5-6 s later, 4 of 4.
+//  2. The private-node SYN gate dropped the service's dial-back reply
+//     (about 50 "SYN rejected: untrusted source" per failed query).
+//  3. The dial spent its SYN retries, and queued duplicate SYNs, while
+//     the key exchange was still running.
+
+import (
+	"bytes"
+	"context"
+	"crypto/ed25519"
+	"errors"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/protocol"
+	"github.com/pilot-protocol/pilotprotocol/pkg/daemon/keyexchange"
+)
+
+// kxProxy forwards UDP frames between a client tunnel and a service
+// tunnel on loopback, so a test can lose or delay what the service sends
+// back, the way a lossy or congested path does. The service sees every
+// client frame as coming from the proxy.
+type kxProxy struct {
+	conn    *net.UDPConn
+	service *net.UDPAddr
+
+	mu            sync.Mutex
+	client        *net.UDPAddr
+	dropFirstPILA bool      // drop the service's first PILA to the client
+	holdUntil     time.Time // buffer service->client frames until then
+	held          [][]byte
+	droppedPILA   int
+}
+
+func newKXProxy(t *testing.T, service *net.UDPAddr) *kxProxy {
+	t.Helper()
+	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatalf("proxy listen: %v", err)
+	}
+	p := &kxProxy{conn: c, service: service}
+	t.Cleanup(func() { c.Close() })
+	go p.run()
+	return p
+}
+
+func (p *kxProxy) addr() *net.UDPAddr { return p.conn.LocalAddr().(*net.UDPAddr) }
+
+// setDropFirstPILA arms dropping the service's next PILA. The run
+// goroutine reads the flag under mu, so tests must set it through here.
+func (p *kxProxy) setDropFirstPILA(v bool) {
+	p.mu.Lock()
+	p.dropFirstPILA = v
+	p.mu.Unlock()
+}
+
+func (p *kxProxy) run() {
+	buf := make([]byte, 65535)
+	for {
+		n, from, err := p.conn.ReadFromUDP(buf)
+		if err != nil {
+			return
+		}
+		frame := append([]byte(nil), buf[:n]...)
+		if from.Port == p.service.Port && from.IP.Equal(p.service.IP) {
+			p.mu.Lock()
+			client := p.client
+			if p.dropFirstPILA && n >= 4 && bytes.Equal(frame[:4], protocol.TunnelMagicAuthEx[:]) {
+				p.dropFirstPILA = false
+				p.droppedPILA++
+				p.mu.Unlock()
+				continue
+			}
+			if time.Now().Before(p.holdUntil) {
+				p.held = append(p.held, frame)
+				p.mu.Unlock()
+				continue
+			}
+			p.mu.Unlock()
+			if client != nil {
+				p.conn.WriteToUDP(frame, client)
+			}
+			continue
+		}
+		p.mu.Lock()
+		p.client = from
+		p.mu.Unlock()
+		p.conn.WriteToUDP(frame, p.service)
+	}
+}
+
+// hold buffers everything the service sends until d has passed, then
+// delivers it in order.
+func (p *kxProxy) hold(d time.Duration) {
+	p.mu.Lock()
+	p.holdUntil = time.Now().Add(d)
+	p.mu.Unlock()
+	time.AfterFunc(d+20*time.Millisecond, func() {
+		p.mu.Lock()
+		held, client := p.held, p.client
+		p.held = nil
+		p.mu.Unlock()
+		for _, f := range held {
+			if client != nil {
+				p.conn.WriteToUDP(f, client)
+			}
+		}
+	})
+}
+
+// newFirstContactPair builds a client and a service daemon on loopback
+// with authenticated identities, the client reaching the service through
+// a kxProxy. No key exchange has happened yet.
+func newFirstContactPair(t *testing.T, clientCfg, serviceCfg Config, clientID, serviceID uint32) (client, service *Daemon, proxy *kxProxy) {
+	t.Helper()
+	client, service = New(clientCfg), New(serviceCfg)
+	keys := map[uint32]ed25519.PublicKey{
+		clientID:  startAuthLoopbackTunnel(t, client.tunnels, clientID),
+		serviceID: startAuthLoopbackTunnel(t, service.tunnels, serviceID),
+	}
+	client.tunnels.SetPeerVerifyFunc(registryKeys(keys))
+	service.tunnels.SetPeerVerifyFunc(registryKeys(keys))
+	client.setNodeID_testhelper(clientID)
+	service.setNodeID_testhelper(serviceID)
+	proxy = newKXProxy(t, tunnelUDPAddr(t, service.tunnels))
+	return client, service, proxy
+}
+
+// TestFirstContactLostKeyReplyRecoversWithKeyRequest is cause 1: the
+// service's one reply to our first PILA is lost. A plain same-key PILA
+// retransmit is answered with silence (the deadlock); the retransmit's
+// key request makes the service send its key again, and the service
+// stops retransmitting once our PILA confirms we have it.
+func TestFirstContactLostKeyReplyRecoversWithKeyRequest(t *testing.T) {
+	t.Parallel()
+	const clientID, serviceID = uint32(0x0C01), uint32(0x0C02)
+	client, service, proxy := newFirstContactPair(t, Config{}, Config{Public: true}, clientID, serviceID)
+	a, b := client.tunnels, service.tunnels
+	proxy.setDropFirstPILA(true)
+
+	a.AddPeer(serviceID, proxy.addr())
+	waitUntil(t, "service installs our key", func() bool { return b.HasCrypto(clientID) })
+	time.Sleep(keyexchange.DuplicateHandshakeDebounce + 100*time.Millisecond)
+	if a.HasCrypto(serviceID) {
+		t.Fatal("setup: the service's reply should have been lost")
+	}
+
+	// The deadlock: the same PILA again is a same-session keepalive to the
+	// service, which stays silent.
+	if err := a.writeFrame(serviceID, proxy.addr(), a.buildAuthKeyExchangeFrame()); err != nil {
+		t.Fatalf("writeFrame: %v", err)
+	}
+	time.Sleep(500 * time.Millisecond)
+	if a.HasCrypto(serviceID) {
+		t.Fatal("a same-key PILA was answered; the deadlock this test pins no longer exists")
+	}
+
+	// Our retransmit, due now, carries the key request.
+	a.kx.InjectPendingRekeyForTest(serviceID, &keyexchange.PendingRekeyState{
+		FirstSentAt: time.Now().Add(-10 * time.Second),
+		LastSentAt:  time.Now().Add(-5 * time.Second),
+		Attempts:    1,
+	})
+	a.rekeyRetransmitTick()
+	waitUntil(t, "client installs the service's key", func() bool { return a.HasCrypto(serviceID) })
+	if got := a.kx.KeyRequestsSent(); got < 1 {
+		t.Fatalf("key requests sent = %d, want >= 1", got)
+	}
+	// Our PILA in answer to the service's key confirms the session, which
+	// ends the service's own retransmits of its reply.
+	waitUntil(t, "service stops retransmitting its key", func() bool { return !b.kx.PendingRekeyHas(clientID) })
+}
+
+// TestFirstContactKeyRequestNotSentWhileSessionUp: a retransmit to a
+// peer we already hold a session key for (an ordinary rekey) carries no
+// key request.
+func TestFirstContactKeyRequestNotSentWhileSessionUp(t *testing.T) {
+	t.Parallel()
+	const clientID, serviceID = uint32(0x0C11), uint32(0x0C12)
+	client, service, proxy := newFirstContactPair(t, Config{}, Config{Public: true}, clientID, serviceID)
+	a, b := client.tunnels, service.tunnels
+
+	a.AddPeer(serviceID, proxy.addr())
+	waitUntil(t, "handshake", func() bool { return a.HasCrypto(serviceID) && b.HasCrypto(clientID) })
+	a.kx.InjectPendingRekeyForTest(serviceID, &keyexchange.PendingRekeyState{
+		FirstSentAt: time.Now().Add(-10 * time.Second),
+		LastSentAt:  time.Now().Add(-5 * time.Second),
+		Attempts:    3,
+	})
+	a.rekeyRetransmitTick()
+	if got := a.kx.KeyRequestsSent(); got != 0 {
+		t.Fatalf("key requests sent with a session up = %d, want 0", got)
+	}
+}
+
+// TestNoKeyFrameFromPeerSendsKeyRequest: an encrypted frame from a peer
+// we hold no key for proves the peer holds a session for us; the rekey
+// request we answer it with carries a key request too, on the first try.
+func TestNoKeyFrameFromPeerSendsKeyRequest(t *testing.T) {
+	t.Parallel()
+	a := NewTunnelManager()
+	startAuthLoopbackTunnel(t, a, 0x0C21)
+	peer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	defer peer.Close()
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	const peerID = uint32(0x0C22)
+	a.mu.Lock()
+	a.peers[peerID] = peerAddr
+	a.mu.Unlock()
+
+	if !a.maybeRequestRekey(peerID, peerAddr) {
+		t.Fatal("rekey request not sent")
+	}
+	var magics [][]byte
+	for {
+		f := readOneFrame(t, peer)
+		if f == nil {
+			break
+		}
+		magics = append(magics, f[:4])
+	}
+	var pila, pilk int
+	for _, m := range magics {
+		switch {
+		case bytes.Equal(m, protocol.TunnelMagicAuthEx[:]):
+			pila++
+		case bytes.Equal(m, protocol.TunnelMagicKeyEx[:]):
+			pilk++
+		}
+	}
+	if pila != 1 || pilk != 1 {
+		t.Fatalf("frames sent: %d PILA, %d PILK; want 1 and 1", pila, pilk)
+	}
+}
+
+// startRouteLoop runs d's packet dispatch the way Start does.
+func startRouteLoop(t *testing.T, d *Daemon) {
+	t.Helper()
+	go d.routeLoop()
+	t.Cleanup(func() { d.stopOnce.Do(func() { close(d.stopCh) }) })
+}
+
+// pendingFrames returns how many frames the tunnel holds for node while
+// it waits for the key exchange.
+func pendingFrames(tm *TunnelManager, node uint32) int {
+	tm.pendMu.Lock()
+	defer tm.pendMu.Unlock()
+	return len(tm.pending[node])
+}
+
+// TestDialWaitsForKeyWithoutQueuingDuplicateSYNs is cause 3: while the
+// key exchange with a slow peer runs, the dial keeps one SYN queued (sent
+// the moment the key arrives) instead of retransmitting into the queue
+// and burning its retry budget, and it completes once the key is in.
+func TestDialWaitsForKeyWithoutQueuingDuplicateSYNs(t *testing.T) {
+	t.Parallel()
+	const clientID, serviceID = uint32(0x0C31), uint32(0x0C32)
+	client, service, proxy := newFirstContactPair(t, Config{Encrypt: true}, Config{Public: true, Encrypt: true}, clientID, serviceID)
+	if _, err := service.ports.Bind(protocol.PortDataExchange); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+	startRouteLoop(t, client)
+	startRouteLoop(t, service)
+
+	// The service's key-exchange reply takes 2.5 s to come back: past the
+	// old direct phase and several SYN retransmits.
+	proxy.hold(2500 * time.Millisecond)
+	client.tunnels.AddPeer(serviceID, proxy.addr())
+
+	type result struct {
+		conn *Connection
+		err  error
+	}
+	done := make(chan result, 1)
+	go func() {
+		c, err := client.DialConnectionContext(context.Background(), protocol.Addr{Node: serviceID}, protocol.PortDataExchange)
+		done <- result{c, err}
+	}()
+
+	time.Sleep(2 * time.Second)
+	if client.tunnels.IsEncrypted(serviceID) {
+		t.Fatal("setup: key exchange should still be held up")
+	}
+	if n := pendingFrames(client.tunnels, serviceID); n != 1 {
+		t.Fatalf("frames queued for the peer during key exchange = %d, want exactly the one SYN", n)
+	}
+	select {
+	case r := <-done:
+		t.Fatalf("dial returned during key exchange: %v", r.err)
+	default:
+	}
+
+	select {
+	case r := <-done:
+		if r.err != nil {
+			t.Fatalf("dial: %v", r.err)
+		}
+		r.conn.Mu.Lock()
+		st := r.conn.State
+		r.conn.Mu.Unlock()
+		if st != StateEstablished {
+			t.Fatalf("state = %v, want ESTABLISHED", st)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("dial did not complete after the key exchange finished")
+	}
+}
+
+// TestDialKeyExchangeErrorWrapsDialTimeout: callers that only check for
+// a dial timeout keep working with the more specific error.
+func TestDialKeyExchangeErrorWrapsDialTimeout(t *testing.T) {
+	t.Parallel()
+	if !errors.Is(ErrDialKeyExchange, protocol.ErrDialTimeout) {
+		t.Fatal("ErrDialKeyExchange must wrap protocol.ErrDialTimeout")
+	}
+	if got := ErrDialKeyExchange.Error(); got != "dial timeout: key exchange with peer did not complete" {
+		t.Fatalf("message = %q", got)
+	}
+}
+
+// TestDialViaRelayWhenKeyArrivedOnlyOverRelay: the relay is the only path
+// a peer has proven when its key came in over the relay and nothing was
+// ever decrypted from its direct address.
+func TestKeyArrivedViaRelayOnly(t *testing.T) {
+	t.Parallel()
+	tm := NewTunnelManager()
+	const peer = uint32(0x0C41)
+	if tm.KeyArrivedViaRelayOnly(peer) {
+		t.Fatal("unknown peer reported relay-only")
+	}
+	tm.keyViaRelay.Store(peer, true)
+	if !tm.KeyArrivedViaRelayOnly(peer) {
+		t.Fatal("key via relay, no direct traffic: want relay-only")
+	}
+	tm.routing.RecordDirectRecv(peer, time.Now())
+	if tm.KeyArrivedViaRelayOnly(peer) {
+		t.Fatal("a direct decrypt proves the direct path")
+	}
+	tm.keyViaRelay.Store(peer, false)
+	tm.routing.RemovePeer(peer)
+	if tm.KeyArrivedViaRelayOnly(peer) {
+		t.Fatal("key via direct: not relay-only")
+	}
+	tm.keyViaRelay.Store(peer, true)
+	tm.RemovePeer(peer)
+	if tm.KeyArrivedViaRelayOnly(peer) {
+		t.Fatal("RemovePeer must forget the key path")
+	}
+}
+
+// --- cause 2: the reply window in the private-node SYN gate ---
+
+func synFrom(peer, self uint32, dstPort uint16) *protocol.Packet {
+	return newStreamSYNPkt(peer, self, 50000, dstPort, 100)
+}
+
+func TestReplyWindowAdmitsDialBackFromContactedPeer(t *testing.T) {
+	t.Parallel()
+	d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: false})
+	d.setNodeID_testhelper(0xABCD0C51)
+	if _, err := d.ports.Bind(protocol.PortDataExchange); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+
+	// Before we contact the peer, its SYN is dropped silently.
+	d.handleStreamPacket(synFrom(peerNode, d.NodeID(), protocol.PortDataExchange))
+	if pkt := readPacket(t, peerConn, 150*time.Millisecond); pkt != nil {
+		t.Fatalf("SYN from a peer we never contacted was answered (flags=%d)", pkt.Flags)
+	}
+	if n := d.ports.TotalActiveConnections(); n != 0 {
+		t.Fatalf("connections = %d, want 0", n)
+	}
+
+	// We dial it (or send it a handshake): its reply dial-back is admitted.
+	d.noteOutboundContact(peerNode)
+	d.handleStreamPacket(synFrom(peerNode, d.NodeID(), protocol.PortDataExchange))
+	pkt := readPacket(t, peerConn, 500*time.Millisecond)
+	if pkt == nil || !pkt.HasFlag(protocol.FlagSYN) || !pkt.HasFlag(protocol.FlagACK) {
+		t.Fatalf("reply SYN from a contacted peer not answered with SYN-ACK: %+v", pkt)
+	}
+	if n := d.ports.TotalActiveConnections(); n != 1 {
+		t.Fatalf("connections = %d, want 1", n)
+	}
+}
+
+func TestReplyWindowAdmitsHandshakeAccept(t *testing.T) {
+	t.Parallel()
+	d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: false})
+	d.setNodeID_testhelper(0xABCD0C52)
+	if _, err := d.ports.Bind(protocol.PortHandshake); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+	d.noteOutboundContact(peerNode)
+	d.handleStreamPacket(synFrom(peerNode, d.NodeID(), protocol.PortHandshake))
+	if pkt := readPacket(t, peerConn, 500*time.Millisecond); pkt == nil || !pkt.HasFlag(protocol.FlagACK) {
+		t.Fatalf("handshake accept from a contacted peer not answered: %+v", pkt)
+	}
+}
+
+func TestReplyWindowOnlyCoversReplyPorts(t *testing.T) {
+	t.Parallel()
+	d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: false})
+	d.setNodeID_testhelper(0xABCD0C53)
+	if _, err := d.ports.Bind(8080); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+	d.noteOutboundContact(peerNode)
+	d.handleStreamPacket(synFrom(peerNode, d.NodeID(), 8080))
+	if pkt := readPacket(t, peerConn, 150*time.Millisecond); pkt != nil {
+		t.Fatalf("SYN to a non-reply port admitted by the reply window (flags=%d)", pkt.Flags)
+	}
+}
+
+func TestReplyWindowExpires(t *testing.T) {
+	t.Parallel()
+	d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: false})
+	d.setNodeID_testhelper(0xABCD0C54)
+	if _, err := d.ports.Bind(protocol.PortDataExchange); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+	d.outbound.note(peerNode, time.Now().Add(-replyWindow-time.Second))
+	d.handleStreamPacket(synFrom(peerNode, d.NodeID(), protocol.PortDataExchange))
+	if pkt := readPacket(t, peerConn, 150*time.Millisecond); pkt != nil {
+		t.Fatalf("SYN admitted after the reply window expired (flags=%d)", pkt.Flags)
+	}
+	d.outbound.prune(time.Now())
+	if n := d.outbound.len(); n != 0 {
+		t.Fatalf("expired entry not pruned: %d left", n)
+	}
+}
+
+func TestReplyWindowTableIsBounded(t *testing.T) {
+	t.Parallel()
+	var o outboundContacts
+	now := time.Now()
+	for i := 0; i < maxReplyWindowPeers; i++ {
+		o.note(uint32(i+1), now)
+	}
+	o.note(0xFFFF0001, now)
+	if o.len() != maxReplyWindowPeers {
+		t.Fatalf("table grew past its bound: %d", o.len())
+	}
+	if o.recent(0xFFFF0001, now) {
+		t.Fatal("a peer past the bound must fall back to the trust check")
+	}
+	// Once old entries expire, new peers fit again.
+	later := now.Add(replyWindow + time.Second)
+	o.note(0xFFFF0002, later)
+	if !o.recent(0xFFFF0002, later) {
+		t.Fatal("expired entries were not pruned to make room")
+	}
+}
+
+// TestDialOpensReplyWindow: a dial records the contact before its first
+// SYN, so a fast service's reply cannot beat it.
+func TestDialOpensReplyWindow(t *testing.T) {
+	t.Parallel()
+	d, peerNode, _ := setupDaemonWithPeer(t, Config{Public: false})
+	d.setNodeID_testhelper(0xABCD0C55)
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	_, _ = d.DialConnectionContext(ctx, protocol.Addr{Node: peerNode}, protocol.PortDataExchange)
+	if !d.replyWindowAdmits(peerNode, protocol.PortDataExchange) {
+		t.Fatal("dialing a peer must open its reply window")
+	}
+}
+
+// TestKeyInstallDoesNotWaitOnRegistryTrustCheck: installing a new
+// peer's key (and flushing the SYN queued behind it) must not wait on a
+// registry round trip. The only trust question on that path is whether
+// the tunnel.established event may name the peer, and local trust
+// answers it.
+func TestKeyInstallDoesNotWaitOnRegistryTrustCheck(t *testing.T) {
+	t.Parallel()
+	var trustCalls sync.Map
+	rc, stop := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		typ, _ := req["type"].(string)
+		n, _ := trustCalls.LoadOrStore(typ, new(int))
+		*(n.(*int))++
+		return map[string]interface{}{"trusted": false}
+	})
+	defer stop()
+	d := New(Config{Encrypt: true})
+	d.regConn.Store(rc)
+	startAuthLoopbackTunnel(t, d.tunnels, 0x0C61)
+
+	const peerID = uint32(0x0C62)
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) { return peerPub, nil })
+	d.tunnels.handleAuthKeyExchange(data, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 9}, false)
+	if !d.tunnels.HasCrypto(peerID) {
+		t.Fatal("key not installed")
+	}
+	trustCalls.Range(func(k, v any) bool {
+		if typ := k.(string); typ != "" && bytes.Contains([]byte(typ), []byte("trust")) {
+			t.Errorf("key install made %d registry %q call(s)", *(v.(*int)), typ)
+		}
+		return true
+	})
+}