diff --git a/cmd/daemon/main.go b/cmd/daemon/main.go
index 45afa015..43b2d088 100644
--- a/cmd/daemon/main.go
+++ b/cmd/daemon/main.go
@@ -83,6 +83,7 @@ func main() {
 	rekeyWhitelist := flag.String("rekey-whitelist", "", "PILOT-345: comma-separated trusted peer node IDs that bypass the tunnel-rekey interval and 4096 cap. Env: PILOT_REKEY_WHITELIST.")
 	timeWait := flag.Duration("time-wait", 0, "TIME_WAIT duration (default 10s)")
 	public := flag.Bool("public", false, "make this node's endpoint publicly visible (default: private)")
+	strictDataplaneTrust := flag.Bool("strict-dataplane-trust", false, "WS1: refuse key-exchange/control-plane interaction with untrusted peers on a private node. Default false (not enforcing, wire-compatible with old agents). Env: PILOT_STRICT_DATAPLANE_TRUST=1.")
 	relayOnly := flag.Bool("relay-only", false, "hide real_addr from peers; reach this node only via beacon-relay path. Privacy stance: peers cannot enumerate this daemon's public IP. Trade-off: relay adds one beacon hop. Default false (current direct-first behavior).")
 	hostname := flag.String("hostname", "", "hostname for discovery (lowercase alphanumeric + hyphens, max 63 chars)")
 	noEcho := flag.Bool("no-echo", false, "disable built-in echo service (port 7)")
@@ -239,6 +240,7 @@ func main() {
 		MaxTotalConnections:   *maxConnsTotal,
 		TimeWaitDuration:      *timeWait,
 		Public:                *public,
+		StrictDataPlaneTrust:  *strictDataplaneTrust || os.Getenv("PILOT_STRICT_DATAPLANE_TRUST") == "1",
 		RelayOnly:             *relayOnly,
 		Hostname:              *hostname,
 		DisableEcho:           *noEcho,
diff --git a/pkg/daemon/beacon_discovery.go b/pkg/daemon/beacon_discovery.go
index 32d3493c..6c331246 100644
--- a/pkg/daemon/beacon_discovery.go
+++ b/pkg/daemon/beacon_discovery.go
@@ -105,7 +105,7 @@ func initialJitter() time.Duration {
 // (We don't run this on every tick — only at first-tick to keep the
 // hot path simple.)
 func (d *Daemon) beaconRefreshLoop() {
-	if d.beaconSelection == nil || d.regConn == nil || d.identity == nil {
+	if d.beaconSelection == nil || d.reg() == nil || d.identity == nil {
 		// Nothing to refresh — single-beacon static config without
 		// identity/registry. Exit cleanly.
 		return
@@ -152,14 +152,14 @@ func (d *Daemon) beaconRefreshTick(firstTick bool) {
 	// equal to interface nil, so fetchBeaconList's `client == nil`
 	// guard wouldn't catch it — and (*Client).Send panics on a nil
 	// receiver. Bail early from this tick if no registry connection.
-	if d.regConn == nil {
+	if d.reg() == nil {
 		if firstTick {
 			slog.Debug("beacon discovery skipped (no registry connection)")
 		}
 		return
 	}
 
-	discovered, err := fetchBeaconList(d.regConn)
+	discovered, err := fetchBeaconList(d.reg())
 	if err != nil {
 		// On the FIRST tick, try the on-disk cache as a fallback —
 		// the registry may have been briefly unreachable at startup.
diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go
index 129a34b6..ccf9b5b8 100644
--- a/pkg/daemon/daemon.go
+++ b/pkg/daemon/daemon.go
@@ -5,6 +5,7 @@ package daemon
 import (
 	"context"
 	"crypto/ed25519"
+	"crypto/sha256"
 	"crypto/subtle"
 	"crypto/tls"
 	"encoding/base64"
@@ -68,6 +69,28 @@ func isRegistryRejectingUsErr(err error) bool {
 	return false
 }
 
+const registryCallDeadline = 8 * time.Second
+
+var errRegistryCallTimedOut = errors.New("registry: call timed out, connection likely half-open")
+
+func withRegistryDeadline(timeout time.Duration, fn func() (map[string]interface{}, error)) (map[string]interface{}, error) {
+	type registryCallResult struct {
+		resp map[string]interface{}
+		err  error
+	}
+	resultCh := make(chan registryCallResult, 1)
+	go func() {
+		resp, err := fn()
+		resultCh <- registryCallResult{resp: resp, err: err}
+	}()
+	select {
+	case r := <-resultCh:
+		return r.resp, r.err
+	case <-time.After(timeout):
+		return nil, errRegistryCallTimedOut
+	}
+}
+
 var (
 	zeroTime     = func() time.Time { return time.Time{} }
 	fixedTimeout = func() time.Time { return time.Now().Add(5 * time.Second) }
@@ -117,7 +140,8 @@ type Config struct {
 	WebhookRetryBackoff time.Duration // initial retry backoff for webhook POSTs (default 1s)
 
 	// Trust
-	TrustAutoApprove bool // automatically approve all incoming handshake requests
+	TrustAutoApprove     bool // automatically approve all incoming handshake requests
+	StrictDataPlaneTrust bool
 
 	// Fleet enrollment
 	AdminToken string   // admin token for network operations (empty = disabled)
@@ -306,7 +330,7 @@ type Daemon struct {
 	// rotation eliminates the shared-buffer hazard without changing the
 	// existing RLock/RUnlock pattern for non-rotating Sign callers.
 	rotateKeyMu sync.Mutex
-	regConn     *registry.Client
+	regConn     atomic.Pointer[registry.Client]
 	tunnels     *TunnelManager
 	ports       *PortManager
 	ipc         *IPCServer
@@ -545,26 +569,28 @@ func (c *Config) timeWaitDuration() time.Duration {
 
 func New(cfg Config) *Daemon {
 	d := &Daemon{
-		config:        cfg,
-		tunnels:       NewTunnelManager(),
-		ports:         NewPortManager(),
-		stopCh:        make(chan struct{}),
-		synTokens:     cfg.synRateLimit(),
-		synLastFill:   time.Now(),
-		perSrcSYN:     make(map[uint32]*srcSYNBucket),
-		epCache:       make(map[uint32]*endpointEntry),
-		resolveCache:  make(map[uint32]*resolveEntry),
-		hostnameCache: make(map[string]*hostnameCacheEntry),
+		config:          cfg,
+		tunnels:         NewTunnelManager(),
+		ports:           NewPortManager(),
+		stopCh:          make(chan struct{}),
+		synTokens:       cfg.synRateLimit(),
+		synLastFill:     time.Now(),
+		perSrcSYN:       make(map[uint32]*srcSYNBucket),
+		epCache:         make(map[uint32]*endpointEntry),
+		resolveCache:    make(map[uint32]*resolveEntry),
+		hostnameCache:   make(map[string]*hostnameCacheEntry),
 		netPolicies:     make(map[uint16][]uint16),
 		lastGaveUpReset: make(map[uint32]time.Time),
-		managed:       make(map[uint16]*ManagedEngine),
-		memberTags:    make(map[uint16][]string),
+		managed:         make(map[uint16]*ManagedEngine),
+		memberTags:      make(map[uint16][]string),
 	}
 	d.ctx, d.cancelCtx = context.WithCancel(context.Background())
 	d.bus = newInProcessBus(d.NodeID)
 	// Event-driven T2 recovery: reset a peer's path the moment the rekey
 	// machinery gives up on it (see onRekeyGaveUp / pathwatch.go).
 	d.tunnels.SetRekeyGaveUpHook(d.onRekeyGaveUp)
+	d.tunnels.SetTrustGate(d.admitDataPlanePeer)
+	d.tunnels.SetPeerTrustFn(d.isTrustedPeer)
 	d.ipc = NewIPCServer(cfg.SocketPath, d)
 	// HandshakeService is wired post-construction by the composition
 	// root via RegisterHandshakeService (T3.3 — handshake plugin moved
@@ -951,87 +977,11 @@ func (d *Daemon) Start() error {
 	//    to do here.
 
 	// 4. Register with the registry (always with client-generated key).
-	// Retry the initial dial with bounded backoff: in real deployments the
-	// registry may not be listening yet at daemon startup (systemd ordering,
-	// DNAT rules that install a beat after the container starts, etc.). A
-	// hard fail here forces the supervisor to restart us; a short internal
-	// retry is cheaper and avoids a "connection refused" flap storm.
-	var rc *registry.Client
-	var err error
-	const maxRegistryDialAttempts = 10
-	// regConnPoolSize is the number of TCP conns the daemon keeps open to
-	// the registry. With a single conn (the historical default) every
-	// regConn.Send serialises on one mutex — issue #93. 4 conns is enough
-	// to absorb the steady-state burst (heartbeat + per-resolve prewarm +
-	// persistHostnameCache fan-out) and small enough that a few hundred
-	// daemons against one registry server stay well under the registry's
-	// per-host backpressure threshold.
-	const regConnPoolSize = 4
-	registryDialBackoff := 500 * time.Millisecond
-	for attempt := 1; attempt <= maxRegistryDialAttempts; attempt++ {
-		if d.config.RegistryTLS {
-			trust := d.config.RegistryTrust
-			if trust == "" {
-				// Back-compat: RegistryTLS=true used to imply pinning;
-				// require fingerprint if the operator didn't pick a
-				// trust store explicitly.
-				trust = "pinned"
-			}
-			switch trust {
-			case "pinned":
-				if d.config.RegistryFingerprint == "" {
-					return fmt.Errorf("registry TLS with -registry-trust=pinned requires RegistryFingerprint")
-				}
-				rc, err = registry.DialTLSPinned(d.config.RegistryAddr, d.config.RegistryFingerprint)
-			case "system":
-				// OS x509 root store — for registry.pilotprotocol.network
-				// served by Let's Encrypt + nginx SNI routing on 443.
-				rc, err = registry.DialTLSPool(d.config.RegistryAddr, &tls.Config{MinVersion: tls.VersionTLS12}, regConnPoolSize)
-			default:
-				return fmt.Errorf("invalid -registry-trust %q: must be 'pinned' or 'system'", trust)
-			}
-		} else {
-			rc, err = registry.DialPool(d.config.RegistryAddr, regConnPoolSize)
-		}
-		if err == nil {
-			break
-		}
-		if attempt == maxRegistryDialAttempts {
-			return fmt.Errorf("registry dial (after %d attempts): %w", attempt, err)
-		}
-		slog.Warn("registry dial failed, retrying",
-			"attempt", attempt, "max", maxRegistryDialAttempts,
-			"backoff", registryDialBackoff, "error", err)
-		time.Sleep(registryDialBackoff)
-		if registryDialBackoff < 5*time.Second {
-			registryDialBackoff *= 2
-		}
-	}
-	d.regConn = rc
-
-	// H3 fix: set signer for authenticated registry operations.
-	// Read d.identity under d.identityMu on every call so RotateKey (or any
-	// future rebind of d.identity) is picked up without re-installing the
-	// signer. Capturing the pointer at SetSigner time would diverge from
-	// node.PublicKey on the registry after a rotation, producing the symptom
-	// "registry: signature verification failed" on every heartbeat.
-	if d.identity != nil {
-		rc.SetSigner(func(challenge string) string {
-			// Hold identityMu.RLock across Sign(): RotateKey zeros the old
-			// PrivateKey buffer in place under identityMu.Lock(). Releasing
-			// the lock before Sign() would let an in-flight signer read the
-			// very bytes RotateKey is concurrently zeroing (use-after-zero
-			// on signing material). RLock and Lock are mutually exclusive,
-			// so signing and zeroing can never overlap.
-			d.identityMu.RLock()
-			defer d.identityMu.RUnlock()
-			cur := d.identity
-			if cur == nil {
-				return ""
-			}
-			return base64.StdEncoding.EncodeToString(cur.Sign([]byte(challenge)))
-		})
+	rc, err := d.dialRegistryClient()
+	if err != nil {
+		return err
 	}
+	d.regConn.Store(rc)
 
 	pubKeyB64 := crypto.EncodePublicKey(d.identity.PublicKey)
 	resp, err := rc.RegisterWithKeyOpts(registry.RegisterOpts{
@@ -1179,7 +1129,7 @@ func (d *Daemon) Start() error {
 
 	// Set node visibility
 	if d.config.Public {
-		if _, err := d.regConn.SetVisibility(d.nodeID, true); err != nil {
+		if _, err := d.reg().SetVisibility(d.nodeID, true); err != nil {
 			slog.Warn("failed to set public visibility", "error", err)
 		} else {
 			slog.Info("node visibility set", "visibility", "public")
@@ -1188,7 +1138,7 @@ func (d *Daemon) Start() error {
 
 	// Set hostname if configured
 	if d.config.Hostname != "" {
-		if _, err := d.regConn.SetHostname(d.nodeID, d.config.Hostname); err != nil {
+		if _, err := d.reg().SetHostname(d.nodeID, d.config.Hostname); err != nil {
 			slog.Warn("failed to set hostname", "hostname", d.config.Hostname, "error", err)
 		} else {
 			slog.Info("hostname set", "hostname", d.config.Hostname)
@@ -1510,7 +1460,7 @@ func (d *Daemon) autoJoinNetworks() {
 		return
 	}
 	for _, netID := range d.config.Networks {
-		_, err := d.regConn.JoinNetwork(d.nodeID, netID, "", 0, d.config.AdminToken)
+		_, err := d.reg().JoinNetwork(d.nodeID, netID, "", 0, d.config.AdminToken)
 		if err != nil {
 			slog.Warn("auto-join failed", "network_id", netID, "error", err)
 			continue
@@ -1622,8 +1572,8 @@ func (d *Daemon) doStop() {
 	// (including persisted deny/grudge lists). The 5-minute heartbeat TTL
 	// reaps truly-dead nodes; users who explicitly want to leave call
 	// `pilotctl deregister` via IPC (CmdDeregister) which is unaffected.
-	if d.regConn != nil {
-		d.regConn.Close()
+	if d.reg() != nil {
+		d.reg().Close()
 	}
 
 	d.stopPolicyRunners()
@@ -1681,7 +1631,7 @@ func (d *Daemon) doStop() {
 
 // startManaged detects managed networks this node belongs to and starts engines.
 func (d *Daemon) startManaged() {
-	resp, err := d.regConn.ListNetworks()
+	resp, err := d.reg().ListNetworks()
 	if err != nil {
 		slog.Debug("managed: cannot list networks", "err", err)
 		return
@@ -1808,7 +1758,10 @@ func (d *Daemon) nodeNetworks() []uint16 {
 // Used by reconcileMembership where stale cache hides genuine join/leave
 // deltas, and by paths that must observe an up-to-the-millisecond list.
 func (d *Daemon) nodeNetworksFresh() []uint16 {
-	resp, err := d.regConn.Lookup(d.NodeID())
+	rc := d.reg()
+	resp, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+		return rc.Lookup(d.NodeID())
+	})
 	if err != nil {
 		return nil
 	}
@@ -1872,7 +1825,7 @@ func (d *Daemon) loadNetworkPolicies() {
 		// Bounded retry so a transient blip on the very first load (when
 		// there is no prior to fall back to) still self-heals quickly.
 		for attempt := 0; attempt < 3; attempt++ {
-			if resp, err = d.regConn.GetNetworkPolicy(netID); err == nil {
+			if resp, err = d.reg().GetNetworkPolicy(netID); err == nil {
 				break
 			}
 			time.Sleep(200 * time.Millisecond)
@@ -2112,10 +2065,10 @@ func (d *Daemon) PublishEvent(topic string, payload map[string]any) {
 func (d *Daemon) AdminToken() string { return d.config.AdminToken }
 
 func (d *Daemon) RegConnListNodes(netID uint16, token string) (map[string]any, error) {
-	if d.regConn == nil {
+	if d.reg() == nil {
 		return nil, fmt.Errorf("registry connection not initialized")
 	}
-	return d.regConn.ListNodes(netID, token)
+	return d.reg().ListNodes(netID, token)
 }
 
 // TrustedPeers returns the trust records held by the registered
@@ -2249,7 +2202,79 @@ func (d *Daemon) TrustAutoApprove() bool { return d.config.TrustAutoApprove }
 // RequestHandshake / RespondHandshake / PollHandshakes against the
 // same client the daemon uses elsewhere — there is no separate
 // connection or auth context.
-func (d *Daemon) RegistryClient() *registry.Client { return d.regConn }
+func (d *Daemon) RegistryClient() *registry.Client { return d.reg() }
+
+func (d *Daemon) reg() *registry.Client {
+	return d.regConn.Load()
+}
+
+func (d *Daemon) dialRegistryClient() (*registry.Client, error) {
+	var rc *registry.Client
+	var err error
+	const maxRegistryDialAttempts = 10
+	const regConnPoolSize = 4
+	registryDialBackoff := 500 * time.Millisecond
+	for attempt := 1; attempt <= maxRegistryDialAttempts; attempt++ {
+		if d.config.RegistryTLS {
+			trust := d.config.RegistryTrust
+			if trust == "" {
+				trust = "pinned"
+			}
+			switch trust {
+			case "pinned":
+				if d.config.RegistryFingerprint == "" {
+					return nil, fmt.Errorf("registry TLS with -registry-trust=pinned requires RegistryFingerprint")
+				}
+				rc, err = registry.DialTLSPinned(d.config.RegistryAddr, d.config.RegistryFingerprint)
+			case "system":
+				rc, err = registry.DialTLSPool(d.config.RegistryAddr, &tls.Config{MinVersion: tls.VersionTLS12}, regConnPoolSize)
+			default:
+				return nil, fmt.Errorf("invalid -registry-trust %q: must be 'pinned' or 'system'", trust)
+			}
+		} else {
+			rc, err = registry.DialPool(d.config.RegistryAddr, regConnPoolSize)
+		}
+		if err == nil {
+			break
+		}
+		if attempt == maxRegistryDialAttempts {
+			return nil, fmt.Errorf("registry dial (after %d attempts): %w", attempt, err)
+		}
+		slog.Warn("registry dial failed, retrying",
+			"attempt", attempt, "max", maxRegistryDialAttempts,
+			"backoff", registryDialBackoff, "error", err)
+		time.Sleep(registryDialBackoff)
+		if registryDialBackoff < 5*time.Second {
+			registryDialBackoff *= 2
+		}
+	}
+
+	if d.identity != nil {
+		rc.SetSigner(func(challenge string) string {
+			d.identityMu.RLock()
+			defer d.identityMu.RUnlock()
+			cur := d.identity
+			if cur == nil {
+				return ""
+			}
+			return base64.StdEncoding.EncodeToString(cur.Sign([]byte(challenge)))
+		})
+	}
+	return rc, nil
+}
+
+func (d *Daemon) forceReconnectRegistry() error {
+	newConn, err := d.dialRegistryClient()
+	if err != nil {
+		return err
+	}
+	old := d.regConn.Swap(newConn)
+	if old != nil {
+		go old.Close()
+	}
+	slog.Warn("registry connection force-reconnected after half-open detection", "addr", d.config.RegistryAddr)
+	return nil
+}
 
 // peerTagsFor returns the merged tag set for a peer as seen by the policy
 // evaluator: policy-runner local tags (assigned via the `tag` action) unioned
@@ -2314,7 +2339,7 @@ func (d *Daemon) GetMemberTags(netID uint16) []string {
 
 // loadPolicyRunners loads expr policies for all joined networks at startup.
 func (d *Daemon) loadPolicyRunners() {
-	resp, err := d.regConn.ListNetworks()
+	resp, err := d.reg().ListNetworks()
 	if err != nil {
 		slog.Debug("policy: cannot list networks", "err", err)
 		return
@@ -2349,7 +2374,7 @@ func (d *Daemon) loadPolicyRunners() {
 		}
 
 		// Fetch the full policy
-		resp, err := d.regConn.GetExprPolicy(netID)
+		resp, err := d.reg().GetExprPolicy(netID)
 		if err != nil {
 			slog.Warn("policy: cannot fetch expr_policy", "network_id", netID, "err", err)
 			continue
@@ -2440,7 +2465,7 @@ func (d *Daemon) RotateKey() (map[string]interface{}, error) {
 	if current == nil {
 		return nil, fmt.Errorf("rotate_key: daemon has no identity")
 	}
-	if d.regConn == nil {
+	if d.reg() == nil {
 		return nil, fmt.Errorf("rotate_key: registry connection unavailable")
 	}
 
@@ -2457,7 +2482,7 @@ func (d *Daemon) RotateKey() (map[string]interface{}, error) {
 	sig := current.Sign([]byte(challenge))
 	sigB64 := base64.StdEncoding.EncodeToString(sig)
 
-	resp, err := d.regConn.RotateKey(nodeID, sigB64, newPubB64)
+	resp, err := d.reg().RotateKey(nodeID, sigB64, newPubB64)
 	if err != nil {
 		return nil, fmt.Errorf("rotate_key: registry: %w", err)
 	}
@@ -2485,7 +2510,7 @@ func (d *Daemon) RotateKey() (map[string]interface{}, error) {
 	// kept for symmetry with the original RotateKey contract. The closure
 	// here holds the RLock across Sign() (same invariant as Start's signer)
 	// so it is safe against a future rotation's in-place key zeroing.
-	d.regConn.SetSigner(func(c string) string {
+	d.reg().SetSigner(func(c string) string {
 		d.identityMu.RLock()
 		defer d.identityMu.RUnlock()
 		cur := d.identity
@@ -2961,6 +2986,30 @@ func (d *Daemon) routePacketWithRecover(pkt *protocol.Packet, from *net.UDPAddr)
 	d.handlePacket(pkt, from)
 }
 
+func redactID(s string) string {
+	sum := sha256.Sum256([]byte(s))
+	return hex.EncodeToString(sum[:8])
+}
+
+func (d *Daemon) isTrustedPeer(srcNode uint32) bool {
+	trusted := d.handshakes != nil && d.handshakes.IsTrusted(srcNode)
+	if !trusted && d.reg() != nil {
+		var err error
+		trusted, err = d.reg().CheckTrust(d.NodeID(), srcNode)
+		if err != nil {
+			slog.Warn("registry trust check failed (data-plane)", "src_node", srcNode, "err", err)
+		}
+	}
+	return trusted
+}
+
+func (d *Daemon) admitDataPlanePeer(srcNode uint32) bool {
+	if !d.config.StrictDataPlaneTrust || d.config.Public {
+		return true
+	}
+	return d.isTrustedPeer(srcNode)
+}
+
 func (d *Daemon) handlePacket(pkt *protocol.Packet, from *net.UDPAddr) {
 	// D14 mitigation: when encryption is enabled, only auto-add peers that have an
 	// established crypto context (proving prior key exchange). This prevents peer table
@@ -2970,7 +3019,7 @@ func (d *Daemon) handlePacket(pkt *protocol.Packet, from *net.UDPAddr) {
 		if !d.config.Encrypt || d.tunnels.HasCrypto(pkt.Src.Node) {
 			d.tunnels.AddPeer(pkt.Src.Node, from)
 			d.publishEvent("tunnel.peer_added", map[string]interface{}{
-				"peer_node_id": pkt.Src.Node, "endpoint": from.String(),
+				"peer_node_id": pkt.Src.Node,
 			})
 		}
 	}
@@ -3040,10 +3089,10 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 		if !d.config.Public {
 			srcNode := pkt.Src.Node
 			trusted := d.handshakes != nil && d.handshakes.IsTrusted(srcNode)
-			if !trusted && d.regConn != nil {
+			if !trusted && d.reg() != nil {
 				// Fall back to registry trust check (covers admin-set trust pairs + shared networks)
 				var err error
-				trusted, err = d.regConn.CheckTrust(d.NodeID(), srcNode)
+				trusted, err = d.reg().CheckTrust(d.NodeID(), srcNode)
 				if err != nil {
 					slog.Warn("registry trust check failed (SYN)", "src_node", srcNode, "err", err)
 				}
@@ -3051,9 +3100,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 			if !trusted {
 				slog.Warn("SYN rejected: untrusted source", "src_node", srcNode, "src_addr", pkt.Src, "dst_port", pkt.DstPort)
 				d.publishEvent("syn.rejected", map[string]interface{}{
-					"src_node_id": srcNode,
-					"src_addr":    pkt.Src.String(),
-					"dst_port":    pkt.DstPort,
+					"dst_port": pkt.DstPort,
 				})
 				return // silent drop — no RST to avoid leaking node existence
 			}
@@ -3088,7 +3135,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 		if !synWhitelisted && !d.allowSYN() {
 			slog.Warn("SYN rate limit exceeded", "src_addr", pkt.Src, "src_port", pkt.SrcPort)
 			d.publishEvent("security.syn_rate_limited", map[string]interface{}{
-				"src_addr": pkt.Src.String(), "src_port": pkt.SrcPort,
+				"src_addr_hash": redactID(pkt.Src.String()), "src_port": pkt.SrcPort,
 			})
 			return // silently drop — don't even RST (avoid amplification)
 		}
@@ -3468,9 +3515,9 @@ func (d *Daemon) handleDatagramPacket(pkt *protocol.Packet) {
 	if !d.config.Public {
 		srcNode := pkt.Src.Node
 		trusted := d.handshakes != nil && d.handshakes.IsTrusted(srcNode)
-		if !trusted && d.regConn != nil {
+		if !trusted && d.reg() != nil {
 			var err error
-			trusted, err = d.regConn.CheckTrust(d.NodeID(), srcNode)
+			trusted, err = d.reg().CheckTrust(d.NodeID(), srcNode)
 			if err != nil {
 				slog.Warn("registry trust check failed (datagram)", "src_node", srcNode, "err", err)
 			}
@@ -3478,9 +3525,7 @@ func (d *Daemon) handleDatagramPacket(pkt *protocol.Packet) {
 		if !trusted {
 			slog.Warn("datagram rejected: untrusted source", "src_node", srcNode, "src_addr", pkt.Src, "dst_port", pkt.DstPort)
 			d.publishEvent("datagram.rejected", map[string]interface{}{
-				"src_node_id": srcNode,
-				"src_addr":    pkt.Src.String(),
-				"dst_port":    pkt.DstPort,
+				"dst_port": pkt.DstPort,
 			})
 			return
 		}
@@ -3517,6 +3562,9 @@ func (d *Daemon) handleControlPacket(pkt *protocol.Packet) {
 		if pkt.HasFlag(protocol.FlagACK) {
 			return
 		}
+		if !d.admitDataPlanePeer(pkt.Src.Node) {
+			return
+		}
 		// Ping request — send pong back
 		pong := &protocol.Packet{
 			Version:  protocol.Version,
@@ -4511,9 +4559,9 @@ func (d *Daemon) broadcastDatagram(netID uint16, srcPort, dstPort uint16, data [
 	var resp map[string]interface{}
 	var err error
 	if adminToken != "" {
-		resp, err = d.regConn.ListNodes(netID, adminToken)
+		resp, err = d.reg().ListNodes(netID, adminToken)
 	} else {
-		resp, err = d.regConn.ListNodes(netID)
+		resp, err = d.reg().ListNodes(netID)
 	}
 	if err != nil {
 		return fmt.Errorf("list nodes for broadcast: %w", err)
@@ -4925,7 +4973,10 @@ func (d *Daemon) ensureTunnel(nodeID uint32) error {
 	if !cached {
 		// Cache miss — resolve from registry
 		var err error
-		resp, err = d.regConn.Resolve(nodeID, d.NodeID())
+		rc := d.reg()
+		resp, err = withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+			return rc.Resolve(nodeID, d.NodeID())
+		})
 		if err != nil {
 			// Registry unreachable — fall back to cached endpoint
 			if ep, ok := d.cachedEndpoint(nodeID); ok {
@@ -5053,12 +5104,24 @@ func (d *Daemon) trustRepublishLoop() {
 		case <-d.stopCh:
 			return
 		case <-ticker.C:
-			if d.regConn == nil {
+			rc := d.reg()
+			if rc == nil {
 				continue
 			}
-			_, err := d.regConn.Heartbeat(d.NodeID())
+			_, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+				return rc.Heartbeat(d.NodeID())
+			})
 			if err != nil {
 				consecutiveFailures++
+				if errors.Is(err, errRegistryCallTimedOut) {
+					slog.Warn("heartbeat timed out — registry connection likely half-open, forcing reconnect",
+						"consecutive_failures", consecutiveFailures, "deadline", registryCallDeadline)
+					if rcErr := d.forceReconnectRegistry(); rcErr != nil {
+						slog.Warn("registry force-reconnect failed", "error", rcErr)
+					} else {
+						consecutiveFailures = HeartbeatReregThresh
+					}
+				}
 				// If the registry rejects our identity (node not found, or a
 				// signature-verification failure because someone else claimed
 				// our node ID) re-register on this cycle instead of waiting
@@ -5134,7 +5197,7 @@ func (d *Daemon) handshakePollLoop() {
 		case <-d.stopCh:
 			return
 		case <-ticker.C:
-			if d.regConn == nil {
+			if d.reg() == nil {
 				continue
 			}
 			d.pollRelayedHandshakes()
@@ -5202,7 +5265,7 @@ func (d *Daemon) reRegister() {
 	d.identityMu.RLock()
 	pubKeyB64 := crypto.EncodePublicKey(d.identity.PublicKey)
 	d.identityMu.RUnlock()
-	resp, err := d.regConn.RegisterWithKeyOpts(registry.RegisterOpts{
+	resp, err := d.reg().RegisterWithKeyOpts(registry.RegisterOpts{
 		ListenAddr: registrationAddr,
 		PublicKey:  pubKeyB64,
 		Owner:      d.config.Owner,
@@ -5258,12 +5321,12 @@ func (d *Daemon) reRegister() {
 
 	// Restore visibility and hostname after re-registration
 	if d.config.Public {
-		if _, err := d.regConn.SetVisibility(nodeID, true); err != nil {
+		if _, err := d.reg().SetVisibility(nodeID, true); err != nil {
 			slog.Warn("re-registration: failed to restore visibility", "error", err)
 		}
 	}
 	if d.config.Hostname != "" {
-		if _, err := d.regConn.SetHostname(nodeID, d.config.Hostname); err != nil {
+		if _, err := d.reg().SetHostname(nodeID, d.config.Hostname); err != nil {
 			slog.Warn("re-registration: failed to restore hostname", "error", err)
 		}
 	}
@@ -5280,7 +5343,7 @@ func (d *Daemon) reRegister() {
 			if d.stopping() {
 				return
 			}
-			if _, err := d.regConn.ReportTrust(nodeID, rec.NodeID); err != nil {
+			if _, err := d.reg().ReportTrust(nodeID, rec.NodeID); err != nil {
 				slog.Debug("re-registration: failed to re-sync trust pair", "peer", rec.NodeID, "error", err)
 			}
 		}
@@ -5347,10 +5410,10 @@ func (d *Daemon) hostnameReannounceLoop() {
 		case <-d.stopCh:
 			return
 		case <-ticker.C:
-			if d.config.Hostname == "" || d.regConn == nil {
+			if d.config.Hostname == "" || d.reg() == nil {
 				continue
 			}
-			if _, err := d.regConn.SetHostname(d.NodeID(), d.config.Hostname); err != nil {
+			if _, err := d.reg().SetHostname(d.NodeID(), d.config.Hostname); err != nil {
 				slog.Debug("hostname reannounce failed", "hostname", d.config.Hostname, "error", err)
 			}
 		}
@@ -5519,10 +5582,13 @@ func (d *Daemon) tryDirectUpgrade(nodeID uint32) {
 		// A relay tunnel established via beacon discovery never populated
 		// the resolve cache. Resolve fresh so we can target the peer's real
 		// address; without this the upgrade can never start.
-		if d.regConn == nil {
+		rc := d.reg()
+		if rc == nil {
 			return
 		}
-		r, err := d.regConn.Resolve(nodeID, d.NodeID())
+		r, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+			return rc.Resolve(nodeID, d.NodeID())
+		})
 		if err != nil {
 			return
 		}
@@ -5622,7 +5688,7 @@ func (d *Daemon) networkSyncLoop() {
 // a snapshot. It does NOT start or stop policy runners or managed
 // engines directly — those are bus subscribers' responsibility (T4.3).
 func (d *Daemon) reconcileMembership() {
-	if d.regConn == nil {
+	if d.reg() == nil {
 		return
 	}
 
@@ -5656,7 +5722,7 @@ func (d *Daemon) reconcileMembership() {
 	// joined-event payloads will then carry network_id only and
 	// subscribers can fall back to lazy fetch.
 	var networkList []interface{}
-	if listResp, err := d.regConn.ListNetworks(); err == nil {
+	if listResp, err := d.reg().ListNetworks(); err == nil {
 		networkList, _ = listResp["networks"].([]interface{})
 	}
 	listByID := make(map[uint16]map[string]interface{}, len(networkList))
@@ -5739,7 +5805,7 @@ func (d *Daemon) buildJoinPayload(netID uint16, n map[string]interface{}) map[st
 			// to re-issue the L8 RPC. Same lazy-fetch the old
 			// syncPolicyRunners did, but moved up into the publisher
 			// so the policy-plugin subscriber stays L8-free.
-			if pResp, err := d.regConn.GetExprPolicy(netID); err == nil {
+			if pResp, err := d.reg().GetExprPolicy(netID); err == nil {
 				if v, ok := pResp["expr_policy"]; ok && v != nil {
 					payload["expr_policy"] = v
 				}
@@ -5765,7 +5831,7 @@ func (d *Daemon) refreshMemberTagsAndDiff(nets []uint16) map[uint16][]string {
 		if netID == 0 {
 			continue
 		}
-		resp, err := d.regConn.GetMemberTags(netID, nodeID)
+		resp, err := d.reg().GetMemberTags(netID, nodeID)
 		if err != nil {
 			continue
 		}
@@ -5938,7 +6004,7 @@ func (d *Daemon) loadNetworkSnapshot() {
 
 // lookupPeerPubKey fetches a peer's Ed25519 public key from the registry.
 func (d *Daemon) lookupPeerPubKey(nodeID uint32) (ed25519.PublicKey, error) {
-	resp, err := d.regConn.Lookup(nodeID)
+	resp, err := d.reg().Lookup(nodeID)
 	if err != nil {
 		return nil, fmt.Errorf("lookup node %d: %w", nodeID, err)
 	}
@@ -5954,7 +6020,7 @@ func (d *Daemon) lookupPeerPubKey(nodeID uint32) (ed25519.PublicKey, error) {
 // pollRelayedHandshakes checks the registry for handshake requests and
 // responses relayed to this node and processes them.
 func (d *Daemon) pollRelayedHandshakes() {
-	resp, err := d.regConn.PollHandshakes(d.NodeID())
+	resp, err := d.reg().PollHandshakes(d.NodeID())
 	if err != nil {
 		slog.Debug("poll handshakes failed", "error", err)
 		return
diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go
index fa1400e5..6da95b00 100644
--- a/pkg/daemon/ipc.go
+++ b/pkg/daemon/ipc.go
@@ -1280,7 +1280,10 @@ func (s *IPCServer) handleResolveHostname(conn *ipcConn, reqID uint64, payload [
 		s.daemon.hostnameCacheMu.RUnlock()
 	}
 
-	result, err := s.daemon.regConn.ResolveHostname(hostname)
+	rc := s.daemon.reg()
+	result, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+		return rc.ResolveHostname(hostname)
+	})
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("resolve_hostname: %v", err))
 		return
@@ -1309,7 +1312,10 @@ func (s *IPCServer) handleResolveHostname(conn *ipcConn, reqID uint64, payload [
 	if nodeIDVal, ok := result["node_id"].(float64); ok {
 		nodeID := uint32(nodeIDVal)
 		go func() {
-			resolveResp, err := s.daemon.regConn.Resolve(nodeID, s.daemon.NodeID())
+			prewarmRC := s.daemon.reg()
+			resolveResp, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) {
+				return prewarmRC.Resolve(nodeID, s.daemon.NodeID())
+			})
 			if err != nil {
 				slog.Debug("hostname resolve prewarm failed", "node_id", nodeID, "err", err)
 				return
@@ -1340,7 +1346,7 @@ func (s *IPCServer) handleSetHostname(conn *ipcConn, reqID uint64, payload []byt
 	s.daemon.addrMu.RLock()
 	prevHostname := s.daemon.config.Hostname
 	s.daemon.addrMu.RUnlock()
-	result, err := s.daemon.regConn.SetHostname(s.daemon.NodeID(), hostname)
+	result, err := s.daemon.reg().SetHostname(s.daemon.NodeID(), hostname)
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("set_hostname: %v", err))
 		return
@@ -1376,7 +1382,7 @@ func (s *IPCServer) handleSetVisibility(conn *ipcConn, reqID uint64, payload []b
 		return
 	}
 	public := payload[0] == 1
-	result, err := s.daemon.regConn.SetVisibility(s.daemon.NodeID(), public)
+	result, err := s.daemon.reg().SetVisibility(s.daemon.NodeID(), public)
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("set_visibility: %v", err))
 		return
@@ -1396,7 +1402,7 @@ func (s *IPCServer) handleSetVisibility(conn *ipcConn, reqID uint64, payload []b
 }
 
 func (s *IPCServer) handleDeregister(conn *ipcConn, reqID uint64) {
-	result, err := s.daemon.regConn.Deregister(s.daemon.NodeID())
+	result, err := s.daemon.reg().Deregister(s.daemon.NodeID())
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("deregister: %v", err))
 		return
@@ -1421,7 +1427,7 @@ func (s *IPCServer) handleSetTags(conn *ipcConn, reqID uint64, payload []byte) {
 		s.sendError(conn, reqID, "set_tags: maximum 3 tags allowed")
 		return
 	}
-	result, err := s.daemon.regConn.SetTags(s.daemon.NodeID(), tags)
+	result, err := s.daemon.reg().SetTags(s.daemon.NodeID(), tags)
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("set_tags: %v", err))
 		return
@@ -1470,7 +1476,7 @@ func (s *IPCServer) handleSubmitBadge(conn *ipcConn, reqID uint64, payload []byt
 		s.sendError(conn, reqID, "submit_badge: badge and badge_sig required")
 		return
 	}
-	if s.daemon.regConn == nil {
+	if s.daemon.reg() == nil {
 		s.sendError(conn, reqID, "submit_badge: registry connection unavailable")
 		return
 	}
@@ -1481,7 +1487,7 @@ func (s *IPCServer) handleSubmitBadge(conn *ipcConn, reqID uint64, payload []byt
 		return
 	}
 	sigB64 := base64.StdEncoding.EncodeToString(sig)
-	result, err := s.daemon.regConn.SubmitBadge(nodeID, req.Badge, req.BadgeSig, sigB64)
+	result, err := s.daemon.reg().SubmitBadge(nodeID, req.Badge, req.BadgeSig, sigB64)
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("submit_badge: %v", err))
 		return
@@ -1519,7 +1525,7 @@ func (s *IPCServer) handleEnrollRecovery(conn *ipcConn, reqID uint64, payload []
 		s.sendError(conn, reqID, fmt.Sprintf("enroll_recovery: bad enrollment: %v", err))
 		return
 	}
-	if s.daemon.regConn == nil {
+	if s.daemon.reg() == nil {
 		s.sendError(conn, reqID, "enroll_recovery: registry connection unavailable")
 		return
 	}
@@ -1530,7 +1536,7 @@ func (s *IPCServer) handleEnrollRecovery(conn *ipcConn, reqID uint64, payload []
 		return
 	}
 	sigB64 := base64.StdEncoding.EncodeToString(sig)
-	result, err := s.daemon.regConn.EnrollRecovery(nodeID, req.Enrollment, req.EnrollmentSig, sigB64)
+	result, err := s.daemon.reg().EnrollRecovery(nodeID, req.Enrollment, req.EnrollmentSig, sigB64)
 	if err != nil {
 		s.sendError(conn, reqID, fmt.Sprintf("enroll_recovery: %v", err))
 		return
@@ -1733,10 +1739,10 @@ func (s *IPCServer) handleVerifyEnvelope(conn *ipcConn, reqID uint64, payload []
 // key_generation are being added to lookup responses in a parallel work
 // stream, so older registries simply omit them.
 func (s *IPCServer) addEnvelopeStanding(resp map[string]interface{}, e reqsig.Envelope) {
-	if s.daemon.regConn == nil {
+	if s.daemon.reg() == nil {
 		return
 	}
-	lk, err := s.daemon.regConn.Lookup(e.Node)
+	lk, err := s.daemon.reg().Lookup(e.Node)
 	if err != nil {
 		return
 	}
@@ -1969,7 +1975,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 
 	switch sub {
 	case SubNetworkList:
-		result, err := s.daemon.regConn.ListNetworks()
+		result, err := s.daemon.reg().ListNetworks()
 		if err != nil {
 			s.sendError(conn, reqID, fmt.Sprintf("network list: %v", err))
 			return
@@ -1988,7 +1994,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 		if len(rest) > 2 {
 			token = string(rest[2:])
 		}
-		result, err := s.daemon.regConn.JoinNetwork(
+		result, err := s.daemon.reg().JoinNetwork(
 			s.daemon.NodeID(), netID, token, 0, s.daemon.config.AdminToken,
 		)
 		if err != nil {
@@ -2025,7 +2031,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 			return
 		}
 		netID := binary.BigEndian.Uint16(rest[0:2])
-		result, err := s.daemon.regConn.LeaveNetwork(
+		result, err := s.daemon.reg().LeaveNetwork(
 			s.daemon.NodeID(), netID, s.daemon.config.AdminToken,
 		)
 		if err != nil {
@@ -2051,7 +2057,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 			return
 		}
 		netID := binary.BigEndian.Uint16(rest[0:2])
-		result, err := s.daemon.regConn.ListNodes(netID, s.daemon.config.AdminToken)
+		result, err := s.daemon.reg().ListNodes(netID, s.daemon.config.AdminToken)
 		if err != nil {
 			s.sendError(conn, reqID, fmt.Sprintf("network members: %v", err))
 			return
@@ -2067,7 +2073,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 		}
 		netID := binary.BigEndian.Uint16(rest[0:2])
 		targetID := binary.BigEndian.Uint32(rest[2:6])
-		result, err := s.daemon.regConn.InviteToNetwork(
+		result, err := s.daemon.reg().InviteToNetwork(
 			netID, s.daemon.NodeID(), targetID, s.daemon.config.AdminToken,
 		)
 		if err != nil {
@@ -2078,7 +2084,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 		s.ipcWriteNetworkOK(conn, reqID, data)
 
 	case SubNetworkPollInvites:
-		result, err := s.daemon.regConn.PollInvites(s.daemon.NodeID())
+		result, err := s.daemon.reg().PollInvites(s.daemon.NodeID())
 		if err != nil {
 			s.sendError(conn, reqID, fmt.Sprintf("network poll-invites: %v", err))
 			return
@@ -2094,7 +2100,7 @@ func (s *IPCServer) handleNetwork(conn *ipcConn, reqID uint64, payload []byte) {
 		}
 		netID := binary.BigEndian.Uint16(rest[0:2])
 		accept := rest[2] == 1
-		result, err := s.daemon.regConn.RespondInvite(
+		result, err := s.daemon.reg().RespondInvite(
 			s.daemon.NodeID(), netID, accept,
 		)
 		if err != nil {
@@ -2353,7 +2359,7 @@ func (s *IPCServer) handleManaged(conn *ipcConn, reqID uint64, payload []byte) {
 
 		switch action {
 		case 0x00: // get
-			resp, err := s.daemon.regConn.GetMemberTags(tagNetID, targetNodeID)
+			resp, err := s.daemon.reg().GetMemberTags(tagNetID, targetNodeID)
 			if err != nil {
 				s.sendError(conn, reqID, fmt.Sprintf("member-tags get: %v", err))
 				return
@@ -2370,7 +2376,7 @@ func (s *IPCServer) handleManaged(conn *ipcConn, reqID uint64, payload []byte) {
 				s.sendError(conn, reqID, fmt.Sprintf("member-tags set: invalid tags JSON: %v", err))
 				return
 			}
-			resp, err := s.daemon.regConn.SetMemberTags(tagNetID, targetNodeID, tags, s.daemon.config.AdminToken)
+			resp, err := s.daemon.reg().SetMemberTags(tagNetID, targetNodeID, tags, s.daemon.config.AdminToken)
 			if err != nil {
 				s.sendError(conn, reqID, fmt.Sprintf("member-tags set: %v", err))
 				return
diff --git a/pkg/daemon/keyexchange/handle.go b/pkg/daemon/keyexchange/handle.go
index 9d79d0bb..2bfc6499 100644
--- a/pkg/daemon/keyexchange/handle.go
+++ b/pkg/daemon/keyexchange/handle.go
@@ -150,12 +150,15 @@ func (m *Manager) HandleAuthFrame(data []byte, from *net.UDPAddr, fromRelay bool
 			slog.Info("encrypted tunnel established", "auth", true,
 				"peer_node_id", peerNodeID, "endpoint", from, "relay", fromRelay)
 		}
-		m.publish("tunnel.established", map[string]any{
-			"peer_node_id":  peerNodeID,
+		authEstablishedEvent := map[string]any{
 			"authenticated": true,
 			"relay":         fromRelay,
 			"rekeyed":       keyChanged,
-		})
+		}
+		if m.trustFn == nil || m.trustFn(peerNodeID) {
+			authEstablishedEvent["peer_node_id"] = peerNodeID
+		}
+		m.publish("tunnel.established", authEstablishedEvent)
 
 		if m.postInstall != nil {
 			m.postInstall(PostInstallEvent{
@@ -295,12 +298,15 @@ func (m *Manager) HandleUnauthFrame(data []byte, from *net.UDPAddr, fromRelay bo
 			slog.Info("encrypted tunnel established",
 				"peer_node_id", peerNodeID, "endpoint", from, "relay", fromRelay)
 		}
-		m.publish("tunnel.established", map[string]any{
-			"peer_node_id":  peerNodeID,
+		unauthEstablishedEvent := map[string]any{
 			"authenticated": false,
 			"relay":         fromRelay,
 			"rekeyed":       keyChanged,
-		})
+		}
+		if m.trustFn == nil || m.trustFn(peerNodeID) {
+			unauthEstablishedEvent["peer_node_id"] = peerNodeID
+		}
+		m.publish("tunnel.established", unauthEstablishedEvent)
 
 		if m.postInstall != nil {
 			m.postInstall(PostInstallEvent{
diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go
index e8d34f86..67ddbe9a 100644
--- a/pkg/daemon/keyexchange/keyexchange.go
+++ b/pkg/daemon/keyexchange/keyexchange.go
@@ -263,6 +263,7 @@ type Manager struct {
 	postInstall PostInstallHook
 	preRetx     PreRetransmitHook
 	onGaveUp    func(nodeID uint32)
+	trustFn     func(nodeID uint32) bool
 }
 
 // New returns a fresh Manager. The Manager installs into store; pass
@@ -315,6 +316,8 @@ func (m *Manager) SetPreRetransmitHook(h PreRetransmitHook) { m.preRetx = h }
 // to the inbound-silence path watchdog. Invoked outside rkPendingMu.
 func (m *Manager) SetOnGaveUpHook(h func(nodeID uint32)) { m.onGaveUp = h }
 
+func (m *Manager) SetTrustFn(f func(nodeID uint32) bool) { m.trustFn = f }
+
 // SetLocalNodeIDFn supplies the closure used to read our own node ID
 // (atomic read living in the daemon).
 func (m *Manager) SetLocalNodeIDFn(f func() uint32) { m.localNodeIDFn = f }
diff --git a/pkg/daemon/managed.go b/pkg/daemon/managed.go
index 3594dec0..5d39bdeb 100644
--- a/pkg/daemon/managed.go
+++ b/pkg/daemon/managed.go
@@ -352,7 +352,7 @@ func (me *ManagedEngine) fetchMembers() ([]uint32, error) {
 	backoff := 1 * time.Second
 
 	for attempt := 0; attempt < maxAttempts; attempt++ {
-		resp, err := me.daemon.regConn.ListNodes(me.netID, me.daemon.config.AdminToken)
+		resp, err := me.daemon.reg().ListNodes(me.netID, me.daemon.config.AdminToken)
 		if err == nil {
 			nodesRaw, ok := resp["nodes"].([]interface{})
 			if !ok {
diff --git a/pkg/daemon/pathwatch.go b/pkg/daemon/pathwatch.go
index 62b7e375..8aabe524 100644
--- a/pkg/daemon/pathwatch.go
+++ b/pkg/daemon/pathwatch.go
@@ -262,7 +262,7 @@ func (d *Daemon) resetPeerPath(nodeID uint32) peerPathReset {
 
 	// Guarded for watchdog-driven callers in minimal configurations
 	// (tests, registry-less setups): the IPC path always has a regConn.
-	if d.regConn == nil {
+	if d.reg() == nil {
 		res.ResolveErr = "no registry connection"
 		return res
 	}
diff --git a/pkg/daemon/rxwatchdog.go b/pkg/daemon/rxwatchdog.go
index 128a4123..c057eefd 100644
--- a/pkg/daemon/rxwatchdog.go
+++ b/pkg/daemon/rxwatchdog.go
@@ -245,7 +245,7 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa
 		// registry. This is what a manual restart effectively did to clear
 		// the partial wedge.
 		d.tunnels.RegisterWithBeacon()
-		if d.regConn != nil {
+		if d.reg() != nil {
 			d.reRegister()
 		}
 		// Reset the dial counter so the next real dial re-tests the path:
diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go
index cec46527..91efc80a 100644
--- a/pkg/daemon/tunnel.go
+++ b/pkg/daemon/tunnel.go
@@ -167,6 +167,9 @@ type TunnelManager struct {
 	// machinery abandons a peer (see SetRekeyGaveUpHook / resetPeerPath).
 	onRekeyGaveUp func(nodeID uint32)
 
+	trustGate   func(nodeID uint32) bool
+	peerTrustFn func(nodeID uint32) bool
+
 	// Event bus — replaces inline tm.webhook.Emit calls. Set via
 	// SetEventBus from daemon during construction. May be nil in
 	// tests; tm.publishEvent handles that. Webhook delivery is a
@@ -323,6 +326,12 @@ func NewTunnelManager() *TunnelManager {
 			fn(nodeID)
 		}
 	})
+	tm.kx.SetTrustFn(func(nodeID uint32) bool {
+		if fn := tm.peerTrustFn; fn != nil {
+			return fn(nodeID)
+		}
+		return true
+	})
 	return tm
 }
 
@@ -333,6 +342,14 @@ func (tm *TunnelManager) SetRekeyGaveUpHook(fn func(nodeID uint32)) {
 	tm.onRekeyGaveUp = fn
 }
 
+func (tm *TunnelManager) SetTrustGate(fn func(nodeID uint32) bool) {
+	tm.trustGate = fn
+}
+
+func (tm *TunnelManager) SetPeerTrustFn(fn func(nodeID uint32) bool) {
+	tm.peerTrustFn = fn
+}
+
 // maybeForceRelayOnRekey is the cross-layer policy bridge between the
 // keyexchange retransmit loop and the routing layer's relay path.
 // Invoked once per peer per pending retransmit, BEFORE the frame goes
@@ -1265,6 +1282,12 @@ func (tm *TunnelManager) handleAuthKeyExchange(data []byte, from *net.UDPAddr, f
 		slog.Debug("auth key exchange rate-limited (source IP)", "from", from)
 		return
 	}
+	if len(data) >= 4 {
+		peerNodeID := binary.BigEndian.Uint32(data[0:4])
+		if fn := tm.trustGate; fn != nil && !fn(peerNodeID) {
+			return
+		}
+	}
 	tm.kx.HandleAuthFrame(data, from, fromRelay)
 }
 
@@ -1289,6 +1312,12 @@ func (tm *TunnelManager) handleKeyExchange(data []byte, from *net.UDPAddr, fromR
 		slog.Debug("key exchange rate-limited (source IP)", "from", from)
 		return
 	}
+	if len(data) >= 4 {
+		peerNodeID := binary.BigEndian.Uint32(data[0:4])
+		if fn := tm.trustGate; fn != nil && !fn(peerNodeID) {
+			return
+		}
+	}
 	tm.kx.HandleUnauthFrame(data, from, fromRelay)
 }
 
@@ -1426,7 +1455,7 @@ func (tm *TunnelManager) handleEncrypted(data []byte, from *net.UDPAddr) {
 		slog.Warn("tunnel nonce replay detected",
 			"peer_node_id", peerNodeID, "counter", res.Counter, "max", res.MaxRecvNonce)
 		tm.publishEvent("security.nonce_replay", map[string]interface{}{
-			"peer_node_id": peerNodeID, "counter": res.Counter,
+			"peer_hash": redactID(fmt.Sprintf("%d", peerNodeID)), "counter": res.Counter,
 		})
 		// ErrReplay means the frame authenticated (valid AEAD) but the
 		// nonce counter was already seen. Two distinct causes share this
@@ -1530,8 +1559,8 @@ func (tm *TunnelManager) handleEncrypted(data []byte, from *net.UDPAddr) {
 		slog.Warn("tunnel: dropping frame with spoofed source node",
 			"authenticated_peer", peerNodeID, "claimed_src", pkt.Src.Node)
 		tm.publishEvent("security.src_spoofed", map[string]interface{}{
-			"authenticated_peer": peerNodeID,
-			"claimed_src":        pkt.Src.Node,
+			"authenticated_peer_hash": redactID(fmt.Sprintf("%d", peerNodeID)),
+			"claimed_src_hash":        redactID(fmt.Sprintf("%d", pkt.Src.Node)),
 		})
 		return
 	}
diff --git a/pkg/daemon/zz_broadcast_test.go b/pkg/daemon/zz_broadcast_test.go
index d3f130a4..4bcab74b 100644
--- a/pkg/daemon/zz_broadcast_test.go
+++ b/pkg/daemon/zz_broadcast_test.go
@@ -48,7 +48,7 @@ func setupBroadcastFixture(t *testing.T, peerCount int) *broadcastFixture {
 	netID := uint16(createResp["network_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 
 	if err := d.tunnels.Listen("127.0.0.1:0"); err != nil {
diff --git a/pkg/daemon/zz_coverage_pkg_daemon_round2_test.go b/pkg/daemon/zz_coverage_pkg_daemon_round2_test.go
index a90354c1..eff4addb 100644
--- a/pkg/daemon/zz_coverage_pkg_daemon_round2_test.go
+++ b/pkg/daemon/zz_coverage_pkg_daemon_round2_test.go
@@ -54,12 +54,12 @@ import (
 
 func registerSelfOnRegistry(t *testing.T, d *Daemon) uint32 {
 	t.Helper()
-	if d.regConn == nil {
+	if d.reg() == nil {
 		t.Fatal("registerSelfOnRegistry: d.regConn must be set first")
 	}
 	id, _ := crypto.GenerateIdentity()
 	d.identity = id
-	resp, err := d.regConn.RegisterWithKey("127.0.0.1:5000", crypto.EncodePublicKey(id.PublicKey), "", nil)
+	resp, err := d.reg().RegisterWithKey("127.0.0.1:5000", crypto.EncodePublicKey(id.PublicKey), "", nil)
 	if err != nil {
 		t.Fatalf("register self: %v", err)
 	}
@@ -67,7 +67,7 @@ func registerSelfOnRegistry(t *testing.T, d *Daemon) uint32 {
 	d.setNodeID_testhelper(nodeID)
 	// Bind the signer so subsequent registry calls that require a signature
 	// (heartbeat, rotate_key, poll_handshakes) succeed.
-	d.regConn.SetSigner(func(challenge string) string {
+	d.reg().SetSigner(func(challenge string) string {
 		d.identityMu.RLock()
 		cur := d.identity
 		d.identityMu.RUnlock()
@@ -161,7 +161,7 @@ func TestPollRelayedHandshakesEmptyNoOp(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	// No panic, no side effect. The empty-list branches are still walked.
@@ -200,7 +200,7 @@ func TestPollRelayedHandshakesWithServiceNoOp(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	svc := &covRecordingHandshakeService{}
@@ -229,7 +229,7 @@ func TestLoadPolicyRunnersNoNetworksNoOp(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	// Nothing to load, but the function still walks the registry response.
@@ -244,7 +244,7 @@ func TestLoadPolicyRunnersSkipsNetworksWithoutPolicy(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	// Create a network without expr policy — loadPolicyRunners must skip it
@@ -268,7 +268,7 @@ func TestRotateKeyHappyPath(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	oldKey := d.identity.PublicKey
@@ -317,7 +317,7 @@ func TestReconcileMembershipPublishesJoinedEvent(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	// Subscribe to the bus BEFORE creating the network so we don't miss
@@ -349,7 +349,7 @@ func TestReconcileMembershipPublishesLeftEvent(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	if _, err := rc.CreateNetwork(selfID, "leave-test-net", "open", "", "admin-token", false); err != nil {
@@ -531,7 +531,7 @@ func TestBeaconRefreshTickFirstTickEmptyList(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.beaconSelection = newBeaconSelectionState([]string{"bootstrap.example:9001"})
 	id, _ := crypto.GenerateIdentity()
 	d.identity = id
@@ -562,7 +562,7 @@ func TestBeaconRefreshLoopOneTickAndExit(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.beaconSelection = newBeaconSelectionState([]string{"127.0.0.1:9001"})
 	id, _ := crypto.GenerateIdentity()
 	d.identity = id
@@ -637,7 +637,7 @@ func TestManagedEngineForceCycleHappyPath(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	// Need a network to fetch members for.
@@ -683,7 +683,7 @@ func TestLoadPolicyRunnersStartsRunnerForNetworkWithPolicy(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	createResp, err := rc.CreateNetwork(selfID, "policy-net", "open", "", "admin-token", false)
@@ -768,7 +768,7 @@ func TestManagedEngineStartBootstrapAndStop(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	createResp, err := rc.CreateNetwork(selfID, "managed-start-net", "open", "", "admin-token", false)
@@ -806,7 +806,7 @@ func TestHandleRotateKeyHappyPath(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	s := d.ipc
@@ -892,7 +892,7 @@ func TestLoadNetworkPoliciesNoNetworksNoOp(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 	d.loadNetworkPolicies()
 }
@@ -909,7 +909,7 @@ func TestReRegisterHappyPath(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	// reRegister runs synchronously and should succeed against a healthy
@@ -1167,7 +1167,7 @@ func TestHandleManagedCycleReturnsResult(t *testing.T) {
 	reg.SetAdminToken("admin-token")
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	selfID := registerSelfOnRegistry(t, d)
 
 	createResp, err := rc.CreateNetwork(selfID, "managed-cycle-net", "open", "", "admin-token", false)
diff --git a/pkg/daemon/zz_daemon_packet_dispatch_test.go b/pkg/daemon/zz_daemon_packet_dispatch_test.go
index ff8d10e0..3a2a44d7 100644
--- a/pkg/daemon/zz_daemon_packet_dispatch_test.go
+++ b/pkg/daemon/zz_daemon_packet_dispatch_test.go
@@ -20,7 +20,6 @@ func newPacketDaemon(t *testing.T, client *registry.Client) (*Daemon, *net.UDPCo
 		nodeID:       42,
 		tunnels:      NewTunnelManager(),
 		ports:        NewPortManager(),
-		regConn:      client,
 		resolveCache: make(map[uint32]*resolveEntry),
 		epCache:      make(map[uint32]*endpointEntry),
 		netPolicies:  make(map[uint16][]uint16),
@@ -31,6 +30,7 @@ func newPacketDaemon(t *testing.T, client *registry.Client) (*Daemon, *net.UDPCo
 		perSrcSYN:    make(map[uint32]*srcSYNBucket),
 		stopCh:       make(chan struct{}),
 	}
+	d.regConn.Store(client)
 	d.ipc = NewIPCServer("", d)
 	if err := d.tunnels.Listen("127.0.0.1:0"); err != nil {
 		t.Fatalf("tunnel listen: %v", err)
diff --git a/pkg/daemon/zz_ensuretunnel_test.go b/pkg/daemon/zz_ensuretunnel_test.go
index 3ea20522..54e5e782 100644
--- a/pkg/daemon/zz_ensuretunnel_test.go
+++ b/pkg/daemon/zz_ensuretunnel_test.go
@@ -16,7 +16,7 @@ import (
 func TestEnsureTunnelAlreadyHasPeerReturnsNilWithoutResolve(t *testing.T) {
 	t.Parallel()
 	d := New(Config{})
-	// regConn stays nil; if ensureTunnel reached resolve it would panic on d.regConn.Resolve.
+	// regConn stays nil; if ensureTunnel reached resolve it would panic on d.reg().Resolve.
 	d.tunnels.AddPeer(42, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9000})
 
 	if err := d.ensureTunnel(42); err != nil {
@@ -80,7 +80,7 @@ func TestEnsureTunnelRegistryFailsFallsBackToCachedEndpoint(t *testing.T) {
 	rc.Close() // closed client → Resolve errors
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	// Prime the endpoint cache so the fallback branch activates.
 	d.cacheEndpoint(123, "127.0.0.1:41414")
@@ -100,7 +100,7 @@ func TestEnsureTunnelRegistryFailsNoCachedEndpointErrors(t *testing.T) {
 	rc.Close() // closed client → Resolve errors
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	err := d.ensureTunnel(456)
 	if err == nil {
@@ -121,7 +121,7 @@ func TestEnsureTunnelRegistryFailsCachedEndpointUnresolvableErrors(t *testing.T)
 	rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	// Invalid UDP address string — triggers net.ResolveUDPAddr error inside fallback.
 	d.cacheEndpoint(321, "not a host:port")
@@ -158,7 +158,7 @@ func TestDialConnectionEnsureTunnelFailurePropagates(t *testing.T) {
 	rc.Close() // closed client → Resolve errors
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	dst := protocol.Addr{Network: 1, Node: 0xDEADBEEF}
 	conn, err := d.DialConnection(dst, 80)
@@ -210,7 +210,7 @@ func TestDialConnectionAllowedPortStillHitsEnsureTunnel(t *testing.T) {
 	rc.Close() // make ensureTunnel fail quickly
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	d.netPolicyMu.Lock()
 	d.netPolicies[11] = []uint16{80}
@@ -236,7 +236,7 @@ func TestDialConnectionEnsureTunnelFailureBoundedLatency(t *testing.T) {
 	rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	done := make(chan error, 1)
 	go func() {
diff --git a/pkg/daemon/zz_info_health_metrics_bug_test.go b/pkg/daemon/zz_info_health_metrics_bug_test.go
index 178b749c..ce5bf2fa 100644
--- a/pkg/daemon/zz_info_health_metrics_bug_test.go
+++ b/pkg/daemon/zz_info_health_metrics_bug_test.go
@@ -31,7 +31,7 @@ func TestInfoZerosWebhookCountersWhenNoManagerRegistered(t *testing.T) {
 	defer rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	t.Cleanup(func() { d.tunnels.Close() })
 
 	atomic.StoreUint64(&d.AcceptQueueDrops, 7)
@@ -55,7 +55,7 @@ func TestInfoSurfacesWebhookCountersFromManager(t *testing.T) {
 	defer rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	t.Cleanup(func() { d.tunnels.Close() })
 
 	wm := &infoFakeWebhookManager{stats: WebhookStats{Dropped: 13, CircuitSkips: 42}}
diff --git a/pkg/daemon/zz_info_snapshot_test.go b/pkg/daemon/zz_info_snapshot_test.go
index 8065325e..c3d789c3 100644
--- a/pkg/daemon/zz_info_snapshot_test.go
+++ b/pkg/daemon/zz_info_snapshot_test.go
@@ -66,7 +66,8 @@ func TestNodeNetworksLookupErrorReturnsNil(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client, config: Config{AdminToken: "tok"}, nodeID: 99}
+	d := &Daemon{config: Config{AdminToken: "tok"}, nodeID: 99}
+	d.regConn.Store(client)
 	if got := d.nodeNetworks(); got != nil {
 		t.Errorf("error from registry must yield nil slice, got %v", got)
 	}
@@ -84,7 +85,8 @@ func TestNodeNetworksParsesFloat64Entries(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client, nodeID: 99}
+	d := &Daemon{nodeID: 99}
+	d.regConn.Store(client)
 	got := d.nodeNetworks()
 	want := []uint16{5, 7, 65535}
 	if len(got) != len(want) {
@@ -104,7 +106,8 @@ func TestNodeNetworksMissingFieldReturnsEmpty(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client, nodeID: 99}
+	d := &Daemon{nodeID: 99}
+	d.regConn.Store(client)
 	got := d.nodeNetworks()
 	if len(got) != 0 {
 		t.Errorf("missing networks field must yield empty, got %v", got)
@@ -122,7 +125,8 @@ func TestLookupPeerPubKeyLookupError(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client}
+	d := &Daemon{}
+	d.regConn.Store(client)
 	key, err := d.lookupPeerPubKey(42)
 	if err == nil {
 		t.Error("expected error on registry lookup failure")
@@ -139,7 +143,8 @@ func TestLookupPeerPubKeyMissingField(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client}
+	d := &Daemon{}
+	d.regConn.Store(client)
 	_, err := d.lookupPeerPubKey(42)
 	if err == nil {
 		t.Error("expected error when public_key field missing")
@@ -155,7 +160,8 @@ func TestLookupPeerPubKeyEmptyString(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client}
+	d := &Daemon{}
+	d.regConn.Store(client)
 	_, err := d.lookupPeerPubKey(42)
 	if err == nil {
 		t.Error("expected error on empty public_key")
@@ -178,7 +184,8 @@ func TestLookupPeerPubKeyHappyPath(t *testing.T) {
 	})
 	defer cleanup()
 
-	d := &Daemon{regConn: client}
+	d := &Daemon{}
+	d.regConn.Store(client)
 	got, err := d.lookupPeerPubKey(42)
 	if err != nil {
 		t.Fatalf("lookupPeerPubKey: %v", err)
@@ -305,7 +312,7 @@ func TestInfoReturnsSnapshotForFreshDaemon(t *testing.T) {
 		Version:  "v0.0.0-test",
 		Encrypt:  false,
 	})
-	d.regConn = client
+	d.regConn.Store(client)
 	d.nodeID = 7
 	d.addr.Network = 0
 	d.addr.Node = 7
@@ -359,7 +366,7 @@ func TestInfoHasIdentityTrueWhenIdentityPathSet(t *testing.T) {
 	})
 	defer cleanup()
 	d := New(Config{IdentityPath: "/does/not/exist/id.json"})
-	d.regConn = client
+	d.regConn.Store(client)
 	info := d.Info()
 	if !info.Identity {
 		t.Error("Identity should be true when IdentityPath is set")
@@ -376,7 +383,7 @@ func TestInfoIncludesNetworkMembershipsSkippingBackbone(t *testing.T) {
 	defer cleanup()
 
 	d := New(Config{})
-	d.regConn = client
+	d.regConn.Store(client)
 	d.nodeID = 12
 	info := d.Info()
 	// Network 0 is backbone → filtered. Expect 2 memberships.
diff --git a/pkg/daemon/zz_ipc_simple_handlers_test.go b/pkg/daemon/zz_ipc_simple_handlers_test.go
index f524ca3c..90e312b1 100644
--- a/pkg/daemon/zz_ipc_simple_handlers_test.go
+++ b/pkg/daemon/zz_ipc_simple_handlers_test.go
@@ -21,7 +21,6 @@ func newSimpleHandlerDaemon(t *testing.T, client *registry.Client) (*Daemon, *IP
 		nodeID:        7,
 		tunnels:       NewTunnelManager(),
 		ports:         NewPortManager(),
-		regConn:       client,
 		resolveCache:  make(map[uint32]*resolveEntry),
 		epCache:       make(map[uint32]*endpointEntry),
 		hostnameCache: make(map[string]*hostnameCacheEntry),
@@ -30,6 +29,7 @@ func newSimpleHandlerDaemon(t *testing.T, client *registry.Client) (*Daemon, *IP
 		memberTags:    make(map[uint16][]string),
 		startTime:     time.Now(),
 	}
+	d.regConn.Store(client)
 	d.ipc = NewIPCServer("", d)
 	return d, d.ipc
 }
diff --git a/pkg/daemon/zz_ipc_test.go b/pkg/daemon/zz_ipc_test.go
index c6d8ee3f..60cc861b 100644
--- a/pkg/daemon/zz_ipc_test.go
+++ b/pkg/daemon/zz_ipc_test.go
@@ -283,7 +283,7 @@ func TestHandleInfoReturnsValidJSON(t *testing.T) {
 	nodeID := uint32(resp["node_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(nodeID)
 
 	s := d.ipc
@@ -327,7 +327,7 @@ func TestHandleHealthReturnsValidJSON(t *testing.T) {
 	nodeID := uint32(resp["node_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(nodeID)
 
 	s := d.ipc
diff --git a/pkg/daemon/zz_miscipc_test.go b/pkg/daemon/zz_miscipc_test.go
index 8d09141b..e7fcef72 100644
--- a/pkg/daemon/zz_miscipc_test.go
+++ b/pkg/daemon/zz_miscipc_test.go
@@ -39,7 +39,7 @@ func miscTestDaemon(t *testing.T) (*Daemon, uint32) {
 	})
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 	d.setNodeID_testhelper(nodeID)
 	t.Cleanup(func() {
diff --git a/pkg/daemon/zz_network_events_test.go b/pkg/daemon/zz_network_events_test.go
index fbf70d05..035d3c34 100644
--- a/pkg/daemon/zz_network_events_test.go
+++ b/pkg/daemon/zz_network_events_test.go
@@ -112,7 +112,7 @@ func TestNetworkJoinedEventEmitted(t *testing.T) {
 	}
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	initBusForTest(d)
 
@@ -174,7 +174,7 @@ func TestNetworkLeftEventEmitted(t *testing.T) {
 	}
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	initBusForTest(d)
 
@@ -240,7 +240,7 @@ func TestNetworkTagsChangedEventEmitted(t *testing.T) {
 	}
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	initBusForTest(d)
 
@@ -336,7 +336,7 @@ func TestNetworkTagsChangedSuppressedWhenNoDelta(t *testing.T) {
 	}
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	initBusForTest(d)
 
diff --git a/pkg/daemon/zz_networkipc_test.go b/pkg/daemon/zz_networkipc_test.go
index 0942a8cd..0e4b78f6 100644
--- a/pkg/daemon/zz_networkipc_test.go
+++ b/pkg/daemon/zz_networkipc_test.go
@@ -36,7 +36,7 @@ func netTestDaemon(t *testing.T) (*Daemon, uint32) {
 	nodeID := uint32(resp["node_id"].(float64))
 
 	d := New(Config{AdminToken: "admin-token"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 	d.setNodeID_testhelper(nodeID)
 	// Wire the signer so registry calls requiring signatures work.
@@ -59,7 +59,7 @@ func TestHandleNetworkListRepliesOK(t *testing.T) {
 	t.Parallel()
 	d, selfID := netTestDaemon(t)
 	// Pre-create one network so the list is non-empty.
-	if _, err := d.regConn.CreateNetwork(selfID, "alpha", "open", "", "admin-token", false); err != nil {
+	if _, err := d.reg().CreateNetwork(selfID, "alpha", "open", "", "admin-token", false); err != nil {
 		t.Fatalf("create network: %v", err)
 	}
 
@@ -93,12 +93,12 @@ func TestHandleNetworkJoinValidRepliesOK(t *testing.T) {
 	// (self-created networks auto-add creator as a member, which makes a later
 	// join fail with "already in network").
 	otherID, _ := crypto.GenerateIdentity()
-	otherResp, err := d.regConn.RegisterWithKey("127.0.0.1:5602", crypto.EncodePublicKey(otherID.PublicKey), "", nil)
+	otherResp, err := d.reg().RegisterWithKey("127.0.0.1:5602", crypto.EncodePublicKey(otherID.PublicKey), "", nil)
 	if err != nil {
 		t.Fatalf("other register: %v", err)
 	}
 	otherNodeID := uint32(otherResp["node_id"].(float64))
-	createResp, err := d.regConn.CreateNetwork(otherNodeID, "joinable", "open", "", "admin-token", false)
+	createResp, err := d.reg().CreateNetwork(otherNodeID, "joinable", "open", "", "admin-token", false)
 	if err != nil {
 		t.Fatalf("create network: %v", err)
 	}
@@ -146,7 +146,7 @@ func TestHandleNetworkJoinBadNetworkIDSendsError(t *testing.T) {
 func TestHandleNetworkLeaveValidRepliesOK(t *testing.T) {
 	t.Parallel()
 	d, selfID := netTestDaemon(t)
-	createResp, err := d.regConn.CreateNetwork(selfID, "toleave", "open", "", "admin-token", false)
+	createResp, err := d.reg().CreateNetwork(selfID, "toleave", "open", "", "admin-token", false)
 	if err != nil {
 		t.Fatalf("create network: %v", err)
 	}
@@ -171,7 +171,7 @@ func TestHandleNetworkLeaveValidRepliesOK(t *testing.T) {
 func TestHandleNetworkMembersValidRepliesOK(t *testing.T) {
 	t.Parallel()
 	d, selfID := netTestDaemon(t)
-	createResp, err := d.regConn.CreateNetwork(selfID, "member-net", "open", "", "admin-token", false)
+	createResp, err := d.reg().CreateNetwork(selfID, "member-net", "open", "", "admin-token", false)
 	if err != nil {
 		t.Fatalf("create network: %v", err)
 	}
@@ -219,7 +219,7 @@ func TestHandleNetworkInviteValidRepliesOK(t *testing.T) {
 	t.Parallel()
 	d, selfID := netTestDaemon(t)
 	// Create an enterprise invite-only network so InviteToNetwork is meaningful.
-	createResp, err := d.regConn.CreateNetwork(selfID, "closed", "invite", "", "admin-token", true)
+	createResp, err := d.reg().CreateNetwork(selfID, "closed", "invite", "", "admin-token", true)
 	if err != nil {
 		t.Fatalf("create network: %v", err)
 	}
@@ -227,7 +227,7 @@ func TestHandleNetworkInviteValidRepliesOK(t *testing.T) {
 
 	// Register a second node to invite.
 	otherID, _ := crypto.GenerateIdentity()
-	otherResp, err := d.regConn.RegisterWithKey("127.0.0.1:5601", crypto.EncodePublicKey(otherID.PublicKey), "", nil)
+	otherResp, err := d.reg().RegisterWithKey("127.0.0.1:5601", crypto.EncodePublicKey(otherID.PublicKey), "", nil)
 	if err != nil {
 		t.Fatalf("other register: %v", err)
 	}
diff --git a/pkg/daemon/zz_pathwatch_test.go b/pkg/daemon/zz_pathwatch_test.go
index b4af5aeb..671efa01 100644
--- a/pkg/daemon/zz_pathwatch_test.go
+++ b/pkg/daemon/zz_pathwatch_test.go
@@ -227,9 +227,6 @@ func TestResetPeerPathPreservesRelayActive(t *testing.T) {
 	}
 }
 
-
-
-
 // TestOnRekeyGaveUpResetsWithCooldown pins the event-driven T2 recovery:
 // the rekey-gave-up hook resets a peer's path, and a per-peer cooldown
 // prevents a persistently-desynced/offline peer from storming resolves.
diff --git a/pkg/daemon/zz_registry_halfopen_test.go b/pkg/daemon/zz_registry_halfopen_test.go
new file mode 100644
index 00000000..651ba2d8
--- /dev/null
+++ b/pkg/daemon/zz_registry_halfopen_test.go
@@ -0,0 +1,128 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+import (
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/crypto"
+	registryclient "github.com/pilot-protocol/common/registry/client"
+)
+
+func newHalfOpenListener(t *testing.T) net.Listener {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				buf := make([]byte, 4096)
+				for {
+					if _, err := c.Read(buf); err != nil {
+						return
+					}
+				}
+			}(conn)
+		}
+	}()
+	return ln
+}
+
+func TestWithRegistryDeadlineTimesOutOnHalfOpenConn(t *testing.T) {
+	t.Parallel()
+	ln := newHalfOpenListener(t)
+	defer ln.Close()
+
+	rc, err := registryclient.Dial(ln.Addr().String())
+	if err != nil {
+		t.Fatalf("dial half-open listener: %v", err)
+	}
+
+	const deadline = 200 * time.Millisecond
+	start := time.Now()
+	_, err = withRegistryDeadline(deadline, func() (map[string]interface{}, error) {
+		return rc.Lookup(1)
+	})
+	elapsed := time.Since(start)
+
+	if !errors.Is(err, errRegistryCallTimedOut) {
+		t.Fatalf("err = %v, want errRegistryCallTimedOut", err)
+	}
+	if elapsed > 2*deadline {
+		t.Fatalf("withRegistryDeadline took %v, want close to the %v deadline", elapsed, deadline)
+	}
+}
+
+func TestForceReconnectRegistryRecoversFromHalfOpenConn(t *testing.T) {
+	t.Parallel()
+	reg, liveRC := startTestRegistry(t)
+	defer reg.Close()
+	defer liveRC.Close()
+
+	deadLn := newHalfOpenListener(t)
+	defer deadLn.Close()
+
+	deadRC, err := registryclient.Dial(deadLn.Addr().String())
+	if err != nil {
+		t.Fatalf("dial half-open listener: %v", err)
+	}
+
+	id, err := crypto.GenerateIdentity()
+	if err != nil {
+		t.Fatalf("gen identity: %v", err)
+	}
+
+	d := New(Config{RegistryAddr: reg.Addr().String()})
+	d.identity = id
+	d.regConn.Store(deadRC)
+
+	deadlineHit := make(chan error, 1)
+	go func() {
+		_, err := withRegistryDeadline(300*time.Millisecond, func() (map[string]interface{}, error) {
+			return deadRC.Lookup(1)
+		})
+		deadlineHit <- err
+	}()
+	select {
+	case err := <-deadlineHit:
+		if !errors.Is(err, errRegistryCallTimedOut) {
+			t.Fatalf("lookup on half-open conn err = %v, want errRegistryCallTimedOut", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("lookup on half-open conn never returned")
+	}
+
+	start := time.Now()
+	if err := d.forceReconnectRegistry(); err != nil {
+		t.Fatalf("forceReconnectRegistry: %v", err)
+	}
+	elapsed := time.Since(start)
+	if elapsed > 5*time.Second {
+		t.Fatalf("forceReconnectRegistry took %v, want a fresh dial to complete quickly", elapsed)
+	}
+
+	newRC := d.reg()
+	if newRC == deadRC {
+		t.Fatal("regConn was not replaced by forceReconnectRegistry")
+	}
+
+	resp, err := newRC.RegisterWithKeyOpts(registryclient.RegisterOpts{
+		ListenAddr: "127.0.0.1:5500",
+		PublicKey:  crypto.EncodePublicKey(id.PublicKey),
+	})
+	if err != nil {
+		t.Fatalf("post-reconnect register failed: %v", err)
+	}
+	if _, ok := resp["node_id"]; !ok {
+		t.Fatalf("register response missing node_id: %v", resp)
+	}
+}
diff --git a/pkg/daemon/zz_reregister_test.go b/pkg/daemon/zz_reregister_test.go
index 9193f8c6..e51ec5a3 100644
--- a/pkg/daemon/zz_reregister_test.go
+++ b/pkg/daemon/zz_reregister_test.go
@@ -34,7 +34,7 @@ func TestReRegisterHappyPathUpdatesNodeIDAndAddr(t *testing.T) {
 	d := New(Config{
 		ListenAddr: "127.0.0.1:5400",
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 
 	d.reRegister()
@@ -63,7 +63,7 @@ func TestReRegisterPublicInvokesSetVisibility(t *testing.T) {
 		ListenAddr: "127.0.0.1:5401",
 		Public:     true,
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 
 	d.reRegister()
@@ -90,7 +90,7 @@ func TestReRegisterWithHostnameInvokesSetHostname(t *testing.T) {
 		ListenAddr: "127.0.0.1:5402",
 		Hostname:   "test-host",
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 
 	d.reRegister()
@@ -112,7 +112,7 @@ func TestReRegisterWithTrustedPeersReSyncs(t *testing.T) {
 	d := New(Config{
 		ListenAddr: "127.0.0.1:5403",
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 	// Install a fake handshake service with a seeded trusted peer to
 	// exercise the trust-resync branch. The real manager moved to
@@ -145,7 +145,7 @@ func TestReRegisterEndpointBranchTaken(t *testing.T) {
 	d := New(Config{
 		Endpoint: "1.2.3.4:4000",
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 
 	d.reRegister()
@@ -165,7 +165,7 @@ func TestReRegisterFailurePreservesNodeID(t *testing.T) {
 		t.Fatalf("gen identity: %v", err)
 	}
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.identity = id
 	// Seed a pre-existing node ID to confirm it isn't clobbered on failure.
 	d.setNodeID_testhelper(0xBEEF1234)
diff --git a/pkg/daemon/zz_retx_test.go b/pkg/daemon/zz_retx_test.go
index 3b096cfe..2efe34e3 100644
--- a/pkg/daemon/zz_retx_test.go
+++ b/pkg/daemon/zz_retx_test.go
@@ -70,7 +70,7 @@ func TestInfoBasicFieldsWithoutRegistry(t *testing.T) {
 	defer rc.Close()
 
 	d := New(Config{Version: "testv1"})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.startTime = time.Now().Add(-2 * time.Second)
 	d.setNodeID_testhelper(0xABCD0001)
 
diff --git a/pkg/daemon/zz_rotate_key_sign_race_test.go b/pkg/daemon/zz_rotate_key_sign_race_test.go
index 9036625c..8ef46891 100644
--- a/pkg/daemon/zz_rotate_key_sign_race_test.go
+++ b/pkg/daemon/zz_rotate_key_sign_race_test.go
@@ -35,7 +35,7 @@ func TestConcurrentRotateKeyAndSign(t *testing.T) {
 	t.Cleanup(func() { rc.Close() })
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	registerSelfOnRegistry(t, d)
 
 	// The exact signer closure the daemon installs in Start() / RotateKey:
diff --git a/pkg/daemon/zz_rx_watchdog_restartloop_test.go b/pkg/daemon/zz_rx_watchdog_restartloop_test.go
index 364f1d6c..04d9eff8 100644
--- a/pkg/daemon/zz_rx_watchdog_restartloop_test.go
+++ b/pkg/daemon/zz_rx_watchdog_restartloop_test.go
@@ -135,9 +135,9 @@ func TestRecentRxWedgeExits_WindowAndMalformed(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "id.rxwedge")
 	content := strings.Join([]string{
 		strconv.FormatInt(now.Add(-1*time.Minute).UnixNano(), 10), // in window
-		"garbage-not-a-number",                                    // skipped
-		strconv.FormatInt(now.Add(-3*time.Hour).UnixNano(), 10),   // outside window
-		"",                                                        // skipped
+		"garbage-not-a-number",                                  // skipped
+		strconv.FormatInt(now.Add(-3*time.Hour).UnixNano(), 10), // outside window
+		"", // skipped
 		strconv.FormatInt(now.Add(-5*time.Minute).UnixNano(), 10), // in window
 	}, "\n")
 	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
diff --git a/pkg/daemon/zz_senddata_test.go b/pkg/daemon/zz_senddata_test.go
index 9cf82771..9d53c86e 100644
--- a/pkg/daemon/zz_senddata_test.go
+++ b/pkg/daemon/zz_senddata_test.go
@@ -447,7 +447,7 @@ func TestObservabilityHeartbeatLoopPublishesAgentHeartbeat(t *testing.T) {
 func TestTrustRepublishLoopFiresAndStops(t *testing.T) {
 	t.Parallel()
 	d := New(Config{KeepaliveInterval: 50 * time.Millisecond})
-	// regConn is nil — each tick hits the `if d.regConn == nil { continue }`
+	// regConn is nil — each tick hits the `if d.reg() == nil { continue }`
 	// branch. We're only verifying the loop body executes without panicking
 	// and exits on stopCh.
 	done := make(chan struct{})
diff --git a/pkg/daemon/zz_sendpath_test.go b/pkg/daemon/zz_sendpath_test.go
index 67351e6b..d47d728d 100644
--- a/pkg/daemon/zz_sendpath_test.go
+++ b/pkg/daemon/zz_sendpath_test.go
@@ -201,7 +201,7 @@ func TestBroadcastDatagramRegistryClosedReturnsError(t *testing.T) {
 	rc.Close() // force ListNodes to fail
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	err := d.broadcastDatagram(5, 1000, 80, []byte("x"), "")
 	if err == nil {
@@ -218,7 +218,7 @@ func TestLookupPeerPubKeyRegistryClosedReturnsError(t *testing.T) {
 	rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	_, err := d.lookupPeerPubKey(42)
 	if err == nil {
@@ -233,7 +233,7 @@ func TestLookupPeerPubKeyUnknownNodeReturnsError(t *testing.T) {
 	defer rc.Close()
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	_, err := d.lookupPeerPubKey(999999) // never registered
 	if err == nil {
@@ -255,7 +255,7 @@ func TestLookupPeerPubKeySuccessReturnsKey(t *testing.T) {
 	nodeID := uint32(resp["node_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 
 	got, err := d.lookupPeerPubKey(nodeID)
 	if err != nil {
diff --git a/pkg/daemon/zz_startmanaged_test.go b/pkg/daemon/zz_startmanaged_test.go
index 23f22ee9..e56337d3 100644
--- a/pkg/daemon/zz_startmanaged_test.go
+++ b/pkg/daemon/zz_startmanaged_test.go
@@ -51,7 +51,7 @@ func TestStartManagedStartsEngineForMemberManagedNetwork(t *testing.T) {
 	netID := uint16(createResp["network_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(nodeID)
 	defer d.stopManaged()
 
@@ -95,7 +95,7 @@ func TestStartManagedSkipsNonMemberManagedNetwork(t *testing.T) {
 	selfNodeID := uint32(selfResp["node_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	defer d.stopManaged()
 
@@ -134,7 +134,7 @@ func TestManagedEngineBootstrapDoesNotDeadlockWithPersist(t *testing.T) {
 	netID := uint16(createResp["network_id"].(float64))
 
 	d := New(Config{})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(nodeID)
 
 	rules := &registry.NetworkRules{Links: 10, Cycle: "24h", Prune: 2, PruneBy: "age", Fill: 2, FillHow: "random"}
@@ -188,7 +188,7 @@ func TestAutoJoinNetworksJoinsConfiguredNetwork(t *testing.T) {
 		AdminToken: "admin-token",
 		Networks:   []uint16{netID},
 	})
-	d.regConn = rc
+	d.regConn.Store(rc)
 	d.setNodeID_testhelper(selfNodeID)
 	d.autoJoinNetworks()
 
diff --git a/pkg/daemon/zz_strict_dataplane_trust_test.go b/pkg/daemon/zz_strict_dataplane_trust_test.go
new file mode 100644
index 00000000..db5eae4a
--- /dev/null
+++ b/pkg/daemon/zz_strict_dataplane_trust_test.go
@@ -0,0 +1,679 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+import (
+	"crypto/ecdh"
+	"crypto/ed25519"
+	"crypto/rand"
+	"encoding/binary"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/protocol"
+	registry "github.com/pilot-protocol/common/registry/client"
+)
+
+func wireTrustGate(d *Daemon) {
+	d.tunnels.SetTrustGate(d.admitDataPlanePeer)
+	d.tunnels.SetPeerTrustFn(d.isTrustedPeer)
+}
+
+func newStrictBusDaemon(t *testing.T, client *registry.Client) (*Daemon, *net.UDPConn) {
+	t.Helper()
+	d := &Daemon{
+		nodeID:       42,
+		tunnels:      NewTunnelManager(),
+		ports:        NewPortManager(),
+		resolveCache: make(map[uint32]*resolveEntry),
+		epCache:      make(map[uint32]*endpointEntry),
+		netPolicies:  make(map[uint16][]uint16),
+		managed:      make(map[uint16]*ManagedEngine),
+		memberTags:   make(map[uint16][]string),
+		synTokens:    DefaultSYNRateLimit,
+		synLastFill:  time.Now(),
+		perSrcSYN:    make(map[uint32]*srcSYNBucket),
+		stopCh:       make(chan struct{}),
+	}
+	d.regConn.Store(client)
+	d.ipc = NewIPCServer("", d)
+	d.bus = newInProcessBus(d.NodeID)
+	d.tunnels.SetEventBus(d.bus)
+	wireTrustGate(d)
+	if err := d.tunnels.Listen("127.0.0.1:0"); err != nil {
+		t.Fatalf("tunnel listen: %v", err)
+	}
+	peer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatalf("peer listen: %v", err)
+	}
+	t.Cleanup(func() {
+		peer.Close()
+		d.tunnels.Close()
+	})
+	return d, peer
+}
+
+func buildAuthKXFrame(t *testing.T, peerNodeID uint32) (data []byte, peerPub ed25519.PublicKey, peerPriv ed25519.PrivateKey, peerX *ecdh.PrivateKey) {
+	t.Helper()
+	var err error
+	peerPub, peerPriv, err = ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("ed25519 keygen: %v", err)
+	}
+	curve := ecdh.X25519()
+	peerX, err = curve.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("x25519 keygen: %v", err)
+	}
+	challenge := make([]byte, 4+4+32)
+	copy(challenge[0:4], []byte("auth"))
+	binary.BigEndian.PutUint32(challenge[4:8], peerNodeID)
+	copy(challenge[8:40], peerX.PublicKey().Bytes())
+	sig := ed25519.Sign(peerPriv, challenge)
+
+	data = make([]byte, 132)
+	binary.BigEndian.PutUint32(data[0:4], peerNodeID)
+	copy(data[4:36], peerX.PublicKey().Bytes())
+	copy(data[36:68], peerPub)
+	copy(data[68:132], sig)
+	return data, peerPub, peerPriv, peerX
+}
+
+func TestStrictDataPlaneTrustOffAuthKeyExchangeStillEstablishes(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = false
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x1001)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerNodeID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) {
+		return peerPub, nil
+	})
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+	readOneFrame(t, peer)
+
+	d.tunnels.handleAuthKeyExchange(data, peerAddr, false)
+
+	if !d.tunnels.HasCrypto(peerNodeID) {
+		t.Fatalf("strict=false must preserve current behavior: untrusted auth KX should still establish crypto")
+	}
+	if f := readOneFrame(t, peer); f == nil {
+		t.Fatalf("strict=false: expected reciprocal key-exchange reply frame")
+	}
+}
+
+func TestStrictDataPlaneTrustOnAuthKeyExchangeDropsUntrusted(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newStrictBusDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = true
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+
+	sub, unsub := d.bus.Subscribe("tunnel.established")
+	defer unsub()
+
+	const peerNodeID = uint32(0x1002)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerNodeID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) {
+		return peerPub, nil
+	})
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+	readOneFrame(t, peer)
+
+	d.tunnels.handleAuthKeyExchange(data, peerAddr, false)
+
+	if d.tunnels.HasCrypto(peerNodeID) {
+		t.Fatalf("strict=true + private + untrusted: no crypto should be installed")
+	}
+	if f := readOneFrame(t, peer); f != nil {
+		t.Fatalf("strict=true + private + untrusted: no reply frame expected, got %x", f)
+	}
+	select {
+	case ev := <-sub:
+		t.Fatalf("strict=true + private + untrusted: no tunnel.established event expected, got %+v", ev)
+	case <-time.After(150 * time.Millisecond):
+	}
+}
+
+func TestStrictDataPlaneTrustOnAuthKeyExchangeAllowsTrustedPeer(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": true}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = true
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x1003)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerNodeID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) {
+		return peerPub, nil
+	})
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+	readOneFrame(t, peer)
+
+	d.tunnels.handleAuthKeyExchange(data, peerAddr, false)
+
+	if !d.tunnels.HasCrypto(peerNodeID) {
+		t.Fatalf("strict=true but trusted peer: crypto should still be established")
+	}
+	if f := readOneFrame(t, peer); f == nil {
+		t.Fatalf("strict=true but trusted peer: expected reciprocal key-exchange reply")
+	}
+}
+
+func TestStrictDataPlaneTrustOnUnauthKeyExchangeDropsUntrusted(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = true
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x1004)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+
+	curve := ecdh.X25519()
+	peerPriv, err := curve.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("peer keygen: %v", err)
+	}
+	data := make([]byte, 36)
+	binary.BigEndian.PutUint32(data[0:4], peerNodeID)
+	copy(data[4:36], peerPriv.PublicKey().Bytes())
+
+	d.tunnels.handleKeyExchange(data, peerAddr, false)
+
+	if d.tunnels.HasCrypto(peerNodeID) {
+		t.Fatalf("strict=true + private + untrusted: no crypto should be installed via unauth KX")
+	}
+	if f := readOneFrame(t, peer); f != nil {
+		t.Fatalf("strict=true + private + untrusted: no reply frame expected, got %x", f)
+	}
+}
+
+func TestStrictDataPlaneTrustOnControlPingDropsUntrusted(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = true
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x2001)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoControl,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  protocol.PortPing,
+		DstPort:  protocol.PortPing,
+		Seq:      1,
+	}
+	d.handleControlPacket(pkt)
+
+	if f := readOneFrame(t, peer); f != nil {
+		t.Fatalf("strict=true + private + untrusted: no pong expected, got %x", f)
+	}
+}
+
+func TestStrictDataPlaneTrustOffControlPingStillReplies(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = false
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x2002)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoControl,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  protocol.PortPing,
+		DstPort:  protocol.PortPing,
+		Seq:      1,
+	}
+	d.handleControlPacket(pkt)
+
+	if f := readOneFrame(t, peer); f == nil {
+		t.Fatalf("strict=false: expected pong reply (current behavior preserved)")
+	}
+}
+
+func TestStrictDataPlaneTrustOnControlPingAllowsTrustedPeer(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": true}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.config.StrictDataPlaneTrust = true
+	wireTrustGate(d)
+
+	const peerNodeID = uint32(0x2003)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoControl,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  protocol.PortPing,
+		DstPort:  protocol.PortPing,
+		Seq:      1,
+	}
+	d.handleControlPacket(pkt)
+
+	if f := readOneFrame(t, peer); f == nil {
+		t.Fatalf("strict=true but trusted peer: expected pong reply")
+	}
+}
+
+func TestSynRejectedEventOmitsPeerPII(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, peer := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.bus = newInProcessBus(d.NodeID)
+
+	sub, unsub := d.bus.Subscribe("syn.rejected")
+	defer unsub()
+
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	const peerNodeID = uint32(3001)
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+	if _, err := d.ports.Bind(5678); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Flags:    protocol.FlagSYN,
+		Protocol: protocol.ProtoStream,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  3000,
+		DstPort:  5678,
+		Seq:      100,
+	}
+	d.handleStreamPacket(pkt)
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["src_addr"]; ok {
+			t.Errorf("syn.rejected event must not carry src_addr, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["src_node_id"]; ok {
+			t.Errorf("syn.rejected event must not carry src_node_id, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["dst_port"]; !ok {
+			t.Errorf("syn.rejected event should still carry dst_port, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected syn.rejected event")
+	}
+}
+
+func TestDatagramRejectedEventOmitsPeerPII(t *testing.T) {
+	t.Parallel()
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"trusted": false}
+	})
+	defer cleanup()
+
+	d, _ := newPacketDaemon(t, client)
+	d.config.Public = false
+	d.bus = newInProcessBus(d.NodeID)
+
+	sub, unsub := d.bus.Subscribe("datagram.rejected")
+	defer unsub()
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoDatagram,
+		Src:      protocol.Addr{Node: 3002},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  100,
+		DstPort:  200,
+		Payload:  []byte("hello"),
+	}
+	d.handleDatagramPacket(pkt)
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["src_addr"]; ok {
+			t.Errorf("datagram.rejected event must not carry src_addr, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["src_node_id"]; ok {
+			t.Errorf("datagram.rejected event must not carry src_node_id, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["dst_port"]; !ok {
+			t.Errorf("datagram.rejected event should still carry dst_port, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected datagram.rejected event")
+	}
+}
+
+func TestTunnelPeerAddedEventOmitsEndpoint(t *testing.T) {
+	t.Parallel()
+	d, peer := newPacketDaemon(t, nil)
+	d.config.Public = true
+	d.config.Encrypt = false
+	d.bus = newInProcessBus(d.NodeID)
+
+	sub, unsub := d.bus.Subscribe("tunnel.peer_added")
+	defer unsub()
+
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	const peerNodeID = uint32(4001)
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoControl,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  100,
+		DstPort:  9999,
+	}
+	d.handlePacket(pkt, peerAddr)
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["endpoint"]; ok {
+			t.Errorf("tunnel.peer_added event must not carry endpoint, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["peer_node_id"]; !ok {
+			t.Errorf("tunnel.peer_added event should still carry peer_node_id, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected tunnel.peer_added event")
+	}
+}
+
+func TestTunnelEstablishedEventScrubsPeerNodeIDForUntrustedPeer(t *testing.T) {
+	t.Parallel()
+	d, peer := newStrictBusDaemon(t, nil)
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+	d.tunnels.SetPeerTrustFn(func(uint32) bool { return false })
+
+	sub, unsub := d.bus.Subscribe("tunnel.established")
+	defer unsub()
+
+	const peerNodeID = uint32(5001)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerNodeID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) {
+		return peerPub, nil
+	})
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+
+	d.tunnels.handleAuthKeyExchange(data, peerAddr, false)
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["peer_node_id"]; ok {
+			t.Errorf("tunnel.established for an untrusted peer must not carry peer_node_id, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected tunnel.established event")
+	}
+}
+
+func TestTunnelEstablishedEventKeepsPeerNodeIDForTrustedPeer(t *testing.T) {
+	t.Parallel()
+	d, peer := newStrictBusDaemon(t, nil)
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+	d.tunnels.SetPeerTrustFn(func(uint32) bool { return true })
+
+	sub, unsub := d.bus.Subscribe("tunnel.established")
+	defer unsub()
+
+	const peerNodeID = uint32(5002)
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	data, peerPub, _, _ := buildAuthKXFrame(t, peerNodeID)
+	d.tunnels.SetPeerVerifyFunc(func(uint32) (ed25519.PublicKey, error) {
+		return peerPub, nil
+	})
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+
+	d.tunnels.handleAuthKeyExchange(data, peerAddr, false)
+
+	select {
+	case ev := <-sub:
+		if got, ok := ev.Payload["peer_node_id"]; !ok || got != peerNodeID {
+			t.Errorf("tunnel.established for a trusted peer should carry peer_node_id, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected tunnel.established event")
+	}
+}
+
+func TestSecurityNonceReplayEventHashesPeerID(t *testing.T) {
+	t.Parallel()
+	d, _ := newStrictBusDaemon(t, nil)
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+
+	curve := ecdh.X25519()
+	peerPriv, err := curve.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("peer keygen: %v", err)
+	}
+	pc, err := d.tunnels.deriveSecret(peerPriv.PublicKey().Bytes())
+	if err != nil {
+		t.Fatalf("deriveSecret: %v", err)
+	}
+	const peerNodeID = uint32(6001)
+	d.tunnels.mu.Lock()
+	d.tunnels.envelope.Install(peerNodeID, pc)
+	d.tunnels.mu.Unlock()
+
+	pc.ReplayMu.Lock()
+	pc.MaxRecvNonce = 5
+	bit := uint64(5) % replayWindowSize
+	pc.ReplayBitmap[bit/64] |= 1 << (bit % 64)
+	pc.ReplayMu.Unlock()
+
+	sub, unsub := d.bus.Subscribe("security.nonce_replay")
+	defer unsub()
+
+	data := make([]byte, 4+12+16)
+	binary.BigEndian.PutUint32(data[0:4], peerNodeID)
+	copy(data[4:8], pc.NoncePrefix[:])
+	binary.BigEndian.PutUint64(data[8:16], 5)
+
+	d.tunnels.handleEncrypted(data, &net.UDPAddr{})
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["peer_node_id"]; ok {
+			t.Errorf("security.nonce_replay must not carry raw peer_node_id, got %+v", ev.Payload)
+		}
+		hash, ok := ev.Payload["peer_hash"].(string)
+		if !ok || hash == "" {
+			t.Errorf("security.nonce_replay should carry a non-empty peer_hash, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected security.nonce_replay event")
+	}
+}
+
+func TestSecuritySrcSpoofedEventHashesPeerID(t *testing.T) {
+	t.Parallel()
+	d, _ := newStrictBusDaemon(t, nil)
+	d.config.Encrypt = true
+	if err := d.tunnels.EnableEncryption(); err != nil {
+		t.Fatalf("EnableEncryption: %v", err)
+	}
+
+	curve := ecdh.X25519()
+	peerPriv, err := curve.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("peer keygen: %v", err)
+	}
+	pc, err := d.tunnels.deriveSecret(peerPriv.PublicKey().Bytes())
+	if err != nil {
+		t.Fatalf("deriveSecret: %v", err)
+	}
+	const peerNodeID = uint32(6002)
+	d.tunnels.mu.Lock()
+	d.tunnels.envelope.Install(peerNodeID, pc)
+	d.tunnels.mu.Unlock()
+
+	pkt := newPacket("spoofed-payload")
+	pkt.Src.Node = 0x99999999
+	plaintext, err := pkt.Marshal()
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	nonce := make([]byte, pc.AEAD.NonceSize())
+	copy(nonce[0:4], pc.NoncePrefix[:])
+	binary.BigEndian.PutUint64(nonce[4:12], 1)
+	aad := make([]byte, 4)
+	binary.BigEndian.PutUint32(aad, peerNodeID)
+	ct := pc.AEAD.Seal(nil, nonce, plaintext, aad)
+
+	data := make([]byte, 4+12+len(ct))
+	binary.BigEndian.PutUint32(data[0:4], peerNodeID)
+	copy(data[4:16], nonce)
+	copy(data[16:], ct)
+
+	sub, unsub := d.bus.Subscribe("security.src_spoofed")
+	defer unsub()
+
+	d.tunnels.handleEncrypted(data, &net.UDPAddr{})
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["authenticated_peer"]; ok {
+			t.Errorf("security.src_spoofed must not carry raw authenticated_peer, got %+v", ev.Payload)
+		}
+		if _, ok := ev.Payload["claimed_src"]; ok {
+			t.Errorf("security.src_spoofed must not carry raw claimed_src, got %+v", ev.Payload)
+		}
+		if h, ok := ev.Payload["authenticated_peer_hash"].(string); !ok || h == "" {
+			t.Errorf("security.src_spoofed should carry authenticated_peer_hash, got %+v", ev.Payload)
+		}
+		if h, ok := ev.Payload["claimed_src_hash"].(string); !ok || h == "" {
+			t.Errorf("security.src_spoofed should carry claimed_src_hash, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected security.src_spoofed event")
+	}
+}
+
+func TestSecuritySynRateLimitedEventHashesSrcAddr(t *testing.T) {
+	t.Parallel()
+	d, peer := newPacketDaemon(t, nil)
+	d.config.Public = true
+	d.bus = newInProcessBus(d.NodeID)
+
+	sub, unsub := d.bus.Subscribe("security.syn_rate_limited")
+	defer unsub()
+
+	d.synTokens = 0
+
+	peerAddr := peer.LocalAddr().(*net.UDPAddr)
+	const peerNodeID = uint32(7001)
+	d.tunnels.AddPeer(peerNodeID, peerAddr)
+	if _, err := d.ports.Bind(5679); err != nil {
+		t.Fatalf("bind: %v", err)
+	}
+
+	pkt := &protocol.Packet{
+		Version:  protocol.Version,
+		Flags:    protocol.FlagSYN,
+		Protocol: protocol.ProtoStream,
+		Src:      protocol.Addr{Node: peerNodeID},
+		Dst:      protocol.Addr{Node: d.NodeID()},
+		SrcPort:  3000,
+		DstPort:  5679,
+		Seq:      100,
+	}
+	d.handleStreamPacket(pkt)
+
+	select {
+	case ev := <-sub:
+		if _, ok := ev.Payload["src_addr"]; ok {
+			t.Errorf("security.syn_rate_limited must not carry raw src_addr, got %+v", ev.Payload)
+		}
+		hash, ok := ev.Payload["src_addr_hash"].(string)
+		if !ok || hash == "" || strings.Contains(hash, ".") {
+			t.Errorf("security.syn_rate_limited should carry a hashed src_addr_hash, got %+v", ev.Payload)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected security.syn_rate_limited event")
+	}
+}
diff --git a/pkg/daemon/zz_wire_helpers_test.go b/pkg/daemon/zz_wire_helpers_test.go
index b2b5a5c3..55b8722e 100644
--- a/pkg/daemon/zz_wire_helpers_test.go
+++ b/pkg/daemon/zz_wire_helpers_test.go
@@ -22,13 +22,13 @@ func newWireDaemon(t *testing.T, client *registry.Client) (*Daemon, *net.UDPConn
 		nodeID:       42,
 		tunnels:      NewTunnelManager(),
 		ports:        NewPortManager(),
-		regConn:      client,
 		resolveCache: make(map[uint32]*resolveEntry),
 		epCache:      make(map[uint32]*endpointEntry),
 		netPolicies:  make(map[uint16][]uint16),
 		managed:      make(map[uint16]*ManagedEngine),
 		memberTags:   make(map[uint16][]string),
 	}
+	d.regConn.Store(client)
 	if err := d.tunnels.Listen("127.0.0.1:0"); err != nil {
 		t.Fatalf("tunnel listen: %v", err)
 	}
@@ -189,17 +189,18 @@ func TestAutoJoinNetworksJoinsEachAndContinuesOnError(t *testing.T) {
 // that only need regConn and should not pay for UDP binds.
 func newWireDaemonBare(t *testing.T, client *registry.Client) *Daemon {
 	t.Helper()
-	return &Daemon{
+	d := &Daemon{
 		nodeID:       42,
 		tunnels:      NewTunnelManager(),
 		ports:        NewPortManager(),
-		regConn:      client,
 		resolveCache: make(map[uint32]*resolveEntry),
 		epCache:      make(map[uint32]*endpointEntry),
 		netPolicies:  make(map[uint16][]uint16),
 		managed:      make(map[uint16]*ManagedEngine),
 		memberTags:   make(map[uint16][]string),
 	}
+	d.regConn.Store(client)
+	return d
 }
 
 // --- sendRST ---------------------------------------------------------------