feat(transport): NAT hole-punch direct-upgrade + streamed send-file + prefer-direct

Make P2P transfers actually go direct (and stay direct) across NAT, and
ship large files reliably.

- daemon.go: rewrite relayProbeLoop → tryDirectUpgrade. The old loop sent
  a one-way SendDirectProbe every 5 min, which a stateful NAT/firewall
  always drops (no conntrack pinhole). Now it fires a beacon-coordinated
  RequestHolePunch to open the pinhole on both NATs, then pushes encrypted
  probes at the peer's REAL address so the peer's ClearRelayOnDirect
  promotes the path. Unpins blackhole-pinned (non-relay-only) peers,
  resolves fresh when uncached, and runs every 15 s (was 5 min).
- tunnel.go: add SendDirectProbeTo — encrypted probe to an explicit real
  address (the upgrade primitive; the stored peers[] entry for a relay
  peer is the beacon placeholder).
- ipc.go: handlePreferDirect — drop tunnel + cached resolution so the next
  dial re-runs resolve + punch; unpin relay.
- pilotctl/main.go: send-file streams by default (TypeFileStream) and
  falls back to single-frame TypeFile when the peer never sends an
  INIT-ACK (back-compat); --no-stream forces legacy; reports
  transport/sha256/throughput. Adds `prefer-direct` command +
  --prefer-direct/--timeout flags.

Verified on Mac↔GCP-VM (true dual-NAT) and a fresh throwaway VM: tunnel
goes relay=False via hole-punch in ~8 s through a default-deny firewall,
holds direct through a 50 MB transfer (no flip), byte-perfect sha256,
~7-15× the relay throughput. Survives a cold restart of both ends.

go.mod points common/dataexchange at branch commits (pseudo-versions)
pending their tagged releases; the version-bump to proper tags happens at
release time (v1.11.1), which is intentionally held for review.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: teovl

cmd/pilotctl/main.go

@@ -1176,10 +1176,11 @@ Flags:
 
 Publish a message to a topic on a remote node.
 `,
-	"send-file": `Usage: pilotctl send-file <address|hostname> <filepath> [--timeout <dur>]
+	"send-file": `Usage: pilotctl send-file <address|hostname> <filepath> [--timeout <dur>] [--prefer-direct]
 
 Send a file to a remote node via the data-exchange stream. Files are
-capped at 256 MiB (the data-exchange frame ceiling).
+capped at 256 MiB (the data-exchange frame ceiling) unless both daemons
+have raised PILOT_DATAEXCHANGE_MAX_FRAME — see the dataexchange package.
 
 Flags:
   --timeout <dur>       give up if the receiver does not ACK within this
@@ -1189,6 +1190,15 @@ Flags:
                         exits with a non-zero code and a clear hint
                         instead of hanging until SO_KEEPALIVE trips
                         (~120s by default on the OS).
+  --prefer-direct       drop the existing tunnel + sticky relay flag for
+                        this peer before dialing, so the daemon retries
+                        a direct UDP path instead of reusing the
+                        beacon-mediated relay tunnel. Useful when ping
+                        works but send-file hangs — typical sign of a
+                        relay path that established once and got stuck.
+                        Best-effort: if the peer is genuinely behind a
+                        symmetric NAT the daemon will still fall back to
+                        relay within the dial retry budget.
 
 What you see during a transfer (TTY only):
   sending <file> to <target>… <Ns>            self-rewriting elapsed line
@@ -1538,6 +1548,8 @@ dispatch:
 		cmdTrust(cmdArgs)
 	case "trusted":
 		cmdTrusted(cmdArgs)
+	case "prefer-direct":
+		cmdPreferDirect(cmdArgs)
 
 	// Networks
 	case "network":
@@ -3710,7 +3722,64 @@ func cmdSendFile(args []string) {
 	// Best-effort: warns on stderr and continues if handshake fails.
 	maybeAutoHandshake(d, target, false)
 
+	// --prefer-direct breaks the daemon out of a stuck-on-relay tunnel
+	// BEFORE we dial port 1001. Without this, a previously-established
+	// relay tunnel is reused and the dial inherits its broken stream
+	// behavior. We send the IPC, log what the daemon reset, and proceed
+	// regardless — an old daemon returns "unknown command" which we
+	// treat as a best-effort hint, not a hard failure.
+	if flagBool(flags, "prefer-direct") {
+		resp, perr := d.PreferDirect(target.Node)
+		switch {
+		case perr != nil && strings.Contains(perr.Error(), "unknown command"):
+			fmt.Fprintln(os.Stderr, sDim("--prefer-direct: daemon does not support it (pre-v1.12.0); proceeding with existing tunnel"))
+		case perr != nil:
+			fmt.Fprintln(os.Stderr, sDim("--prefer-direct: "+perr.Error()+" (continuing)"))
+		default:
+			had, _ := resp["had_tunnel"].(bool)
+			wasActive, _ := resp["was_relay_active"].(bool)
+			wasPinned, _ := resp["was_relay_pinned"].(bool)
+			fmt.Fprintln(os.Stderr, sDim(fmt.Sprintf("--prefer-direct: tunnel=%v relay_was_active=%v relay_was_pinned=%v",
+				had, wasActive, wasPinned)))
+		}
+	}
+
 	filePath := pos[1]
+	filename := filepath.Base(filePath)
+
+	fi, err := os.Stat(filePath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			fatalCode("not_found", "file not found: %s", filePath)
+		}
+		if os.IsPermission(err) {
+			fatalCode("internal", "permission denied: %s", filePath)
+		}
+		fatalCode("internal", "stat file: %v", err)
+	}
+	if fi.IsDir() {
+		fatalCode("invalid_argument", "%s is a directory, not a file", filePath)
+	}
+	size := fi.Size()
+
+	// Streamed transfer (default): chunked, ACK'd, resumable, end-to-end
+	// SHA-256 verified — no per-frame size cap, and big files no longer
+	// collapse into one giant frame that stalls over relay (or over a
+	// direct link that flips to relay under one-way load). Falls back to
+	// the single-frame TypeFile path when the receiver is too old to
+	// understand TypeFileStream (it never sends an INIT-ACK).
+	if !flagBool(flags, "no-stream") {
+		if res, serr := streamSendFile(d, target, filePath, filename, size, timeout); serr == nil {
+			outputOK(res)
+			return
+		} else if !errors.Is(serr, dataexchange.ErrStreamUnsupported) {
+			fatalHint("connection_failed",
+				"check reachability: pilotctl ping "+target.String()+" · for very large/slow links raise --timeout",
+				"streamed send-file failed: %v", serr)
+		}
+		fmt.Fprintln(os.Stderr, sDim("receiver does not support streamed transfer (pre-v1.12.0); falling back to single-frame TypeFile"))
+	}
+
 	data, err := os.ReadFile(filePath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -3725,14 +3794,12 @@ func cmdSendFile(args []string) {
 	// Reject files that would exceed the data-exchange frame cap before
 	// opening the connection — keeps the failure path clean and avoids
 	// streaming a quarter-gigabyte just to have the receiver close.
-	if len(data) > dataexchange.MaxFrameSize {
+	if uint32(len(data)) > dataexchange.MaxFrameSize {
 		fatalCode("invalid_argument",
-			"file too large: %d bytes (max %d). The chunked-streaming protocol planned in docs/PROPOSAL-reliable-file-transfer.md will lift this; until then split the file or compress it.",
+			"file too large: %d bytes (max %d) for the legacy single-frame path. Use the default streamed transfer (omit --no-stream) against a v1.12.0+ receiver.",
 			len(data), dataexchange.MaxFrameSize)
 	}
 
-	filename := filepath.Base(filePath)
-
 	client, err := dataexchange.Dial(d, target)
 	if err != nil {
 		hint := classifyDaemonError(err)
@@ -3818,6 +3885,54 @@ func cmdSendFile(args []string) {
 	outputOK(result)
 }
 
+// streamSendFile transfers filePath with the chunked, ACK'd, resumable
+// TypeFileStream protocol. It returns a result map on success; the sentinel
+// dataexchange.ErrStreamUnsupported tells the caller to fall back to the
+// single-frame TypeFile path (the receiver is too old). timeout bounds the
+// wait for any single ACK and for the receiver's final verification.
+func streamSendFile(d *driver.Driver, target protocol.Addr, filePath, filename string, size int64, timeout time.Duration) (map[string]interface{}, error) {
+	client, err := dataexchange.Dial(d, target)
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+
+	f, err := os.Open(filePath)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	stop := startWaitProgress(fmt.Sprintf("streaming %s to %s", filename, target))
+	start := time.Now()
+	res, serr := client.SendFileStream(filename, f, size, timeout)
+	stop()
+	if serr != nil {
+		return nil, serr
+	}
+	if !res.OK {
+		return nil, fmt.Errorf("receiver rejected file: %s", res.Message)
+	}
+
+	elapsed := time.Since(start)
+	mbps := 0.0
+	if elapsed > 0 {
+		mbps = (float64(res.TotalBytes) * 8.0) / (1e6 * elapsed.Seconds())
+	}
+	return map[string]interface{}{
+		"filename":        filename,
+		"bytes":           res.TotalBytes,
+		"bytes_sent":      res.BytesSent,
+		"bytes_resumed":   res.BytesResumed,
+		"sha256":          res.Sha256,
+		"destination":     target.String(),
+		"elapsed_ms":      elapsed.Milliseconds(),
+		"throughput_mbps": mbps,
+		"transport":       "filestream",
+		"verified":        res.OK,
+	}, nil
+}
+
 func cmdSendMessage(args []string) {
 	flags, pos := parseFlags(args)
 	if len(pos) < 1 {
@@ -4303,6 +4418,44 @@ func cmdReject(args []string) {
 	}
 }
 
+// cmdPreferDirect drops the daemon's tunnel + sticky routing state for a
+// peer so the next dial retries a fresh direct UDP path.
+//
+// Use case: ping <peer> works (the small UDP fits through the beacon
+// relay just fine) but send-file <peer> hangs ~120s and EOFs — symptom
+// of a relay-mediated tunnel that established once and got stuck for
+// stream traffic. Calling prefer-direct + retrying the dial routes the
+// next attempt through ensureTunnel's resolve-and-punch path, which
+// usually re-establishes a working direct UDP path.
+func cmdPreferDirect(args []string) {
+	if len(args) < 1 {
+		fatalCode("invalid_argument", "usage: pilotctl prefer-direct <node_id|address|hostname>")
+	}
+	d := connectDriver()
+	defer d.Close()
+
+	nodeID := resolveToNodeID(d, args[0])
+	resp, err := d.PreferDirect(nodeID)
+	if err != nil {
+		if strings.Contains(err.Error(), "unknown command") {
+			fatalHint("not_implemented",
+				"upgrade the daemon: brew upgrade pilotprotocol  (or re-run install.sh)",
+				"daemon does not support prefer-direct (pre-v1.12.0)")
+		}
+		fatalCode("connection_failed", "prefer-direct: %v", err)
+	}
+	if jsonOutput {
+		outputOK(resp)
+		return
+	}
+	had, _ := resp["had_tunnel"].(bool)
+	wasActive, _ := resp["was_relay_active"].(bool)
+	wasPinned, _ := resp["was_relay_pinned"].(bool)
+	fmt.Printf("reset routing state for node %d\n", nodeID)
+	fmt.Println(sDim(fmt.Sprintf("  tunnel was up: %v · relay was active: %v · relay was pinned: %v", had, wasActive, wasPinned)))
+	fmt.Println(sDim("  next dial will re-resolve from registry and prefer direct; falls back to relay if direct still fails"))
+}
+
 func cmdUntrust(args []string) {
 	if len(args) < 1 {
 		fatalCode("invalid_argument", "usage: pilotctl untrust <node_id|address|hostname>")

go.mod

@@ -6,8 +6,8 @@ require (
 	github.com/coder/websocket v1.8.14
 	github.com/pilot-protocol/app-store v1.0.1-beta.1.0.20260609061942-8852c785a264
 	github.com/pilot-protocol/beacon v0.2.6
-	github.com/pilot-protocol/common v0.4.8
-	github.com/pilot-protocol/dataexchange v0.2.0
+	github.com/pilot-protocol/common v0.4.9-0.20260615113553-d5cbbfb3e5b6
+	github.com/pilot-protocol/dataexchange v0.2.1-beta.1.0.20260615113607-fac933edea98
 	github.com/pilot-protocol/eventstream v0.2.2
 	github.com/pilot-protocol/handshake v0.2.1
 	github.com/pilot-protocol/nameserver v0.2.1

go.sum

@@ -14,8 +14,12 @@ github.com/pilot-protocol/beacon v0.2.6 h1:grxwaVyPRUT0W6coyjYfNkO0rpzOIrwrKn94S
 github.com/pilot-protocol/beacon v0.2.6/go.mod h1:I/UhEv097g1z/qtAVDZbEhf3R5tzM0Dp71vGHah52A4=
 github.com/pilot-protocol/common v0.4.8 h1:eS2Bc+XcZWJ/qhwwOZbXwIWhtNdOijuoEp716kQE+/c=
 github.com/pilot-protocol/common v0.4.8/go.mod h1:yrAwPXGVMbXU+SADvOCmbdXjK/wJ3uA0KshyLvRlej4=
+github.com/pilot-protocol/common v0.4.9-0.20260615113553-d5cbbfb3e5b6 h1:Us3qSMPTBHPDQXFPY07BoUanriw1rVzS6SAHcbddqzY=
+github.com/pilot-protocol/common v0.4.9-0.20260615113553-d5cbbfb3e5b6/go.mod h1:yrAwPXGVMbXU+SADvOCmbdXjK/wJ3uA0KshyLvRlej4=
 github.com/pilot-protocol/dataexchange v0.2.0 h1:ldE6AyrES1uvdnn1NBl0KZ7C+SSWNtmeHHU3CQhwSCo=
 github.com/pilot-protocol/dataexchange v0.2.0/go.mod h1:JVy2+hr/IjzMPshxjExbGO/4SbJTs7ZJ7iYvT/ODF3Q=
+github.com/pilot-protocol/dataexchange v0.2.1-beta.1.0.20260615113607-fac933edea98 h1:Bqgnf4CZC7aZJyDzz/E7agwXotArJg2FvFlNDqouhLo=
+github.com/pilot-protocol/dataexchange v0.2.1-beta.1.0.20260615113607-fac933edea98/go.mod h1:tM9eyyruBdnxhhUtViasUjnAElwF/r5PQvCYKLdlTLY=
 github.com/pilot-protocol/eventstream v0.2.2 h1:E0IjveK7K+dsIbE/5hD3N821FkHzxVsx1tiAORMzt8k=
 github.com/pilot-protocol/eventstream v0.2.2/go.mod h1:gUjoMEItW1SRJYEq39VlcIeDe2LcE5B18/4bcaUJNrs=
 github.com/pilot-protocol/handshake v0.2.0 h1:uLeV8iNHcsHbcVH+GZ9p7uuIbObA8BReDByF5XGjzB8=

pkg/daemon/daemon.go

@@ -196,8 +196,13 @@ const (
 	MaxZeroWindowProbes = 30
 )
 
-// RelayProbeInterval is how often we probe relay-flagged peers for direct connectivity.
-const RelayProbeInterval = 5 * time.Minute
+// RelayProbeInterval is how often we attempt to upgrade relay-flagged peers
+// to a direct path via coordinated hole-punching. Was 5m (far too slow — a
+// peer stuck on relay stayed there for minutes); tightened to 15s so a peer
+// that becomes direct-capable (or one that briefly flipped to relay under
+// load) is re-punched promptly. The punch + probe is cheap and only runs for
+// peers still flagged relay, so it self-limits once a peer is promoted.
+const RelayProbeInterval = 15 * time.Second
 
 // EndpointCacheTTL is how long a cached endpoint is considered fresh.
 // After this, the entry is stale but still usable as a fallback.
@@ -5204,32 +5209,89 @@ func (d *Daemon) relayProbeLoop() {
 		case <-d.stopCh:
 			return
 		case <-ticker.C:
-			relayPeers := d.tunnels.RelayPeerIDs()
-			for _, nodeID := range relayPeers {
-				// P1-010 fix: send a targeted direct probe without flipping
-				// the relay flag. If the peer's direct path has recovered, the
-				// response will arrive on tm.sock from their real address;
-				// handleEncrypted auto-clears relay mode on a successful
-				// direct decrypt. Concurrent traffic (key exchange replies,
-				// retransmits) continues going via relay during the probe.
-				probe := &protocol.Packet{
-					Version:  protocol.Version,
-					Flags:    protocol.FlagACK,
-					Protocol: protocol.ProtoControl,
-					Src:      d.Addr(),
-					Dst:      protocol.Addr{Network: 0, Node: nodeID},
-					SrcPort:  protocol.PortPing,
-					DstPort:  protocol.PortPing,
-					Seq:      1,
-				}
-				if err := d.tunnels.SendDirectProbe(nodeID, probe); err != nil {
-					slog.Debug("relay direct-probe skipped", "node_id", nodeID, "error", err)
-				}
+			for _, nodeID := range d.tunnels.RelayPeerIDs() {
+				go d.tryDirectUpgrade(nodeID)
 			}
 		}
 	}
 }
 
+// tryDirectUpgrade attempts to promote a relay-flagged peer to a direct path
+// via coordinated NAT hole-punching, then verifies on the rig that the path
+// actually carries traffic.
+//
+// Why this exists: the old relayProbeLoop sent a single one-way SendDirectProbe
+// and assumed the direct path "had recovered". Through a stateful firewall/NAT
+// a one-way probe is dropped — there is no conntrack pinhole until BOTH peers
+// send. So a relay tunnel never upgraded unless the peer was already publicly
+// reachable. Here we (1) coordinate a simultaneous punch via the beacon to
+// open the pinhole on both NATs, then (2) push several encrypted probes at the
+// peer's REAL address (not the beacon placeholder) so the peer's
+// ClearRelayOnDirect promotes the path (DirectClearsRequired=3 direct decrypts).
+func (d *Daemon) tryDirectUpgrade(nodeID uint32) {
+	// Only act on peers we have authoritative resolve info for. Punching a
+	// relay-only peer would leak its real IP via the beacon PunchCommand
+	// (see establishConnection), so require resolve info that says the peer
+	// is NOT relay-only before we punch or probe.
+	resp, ok := d.cachedResolve(nodeID)
+	if !ok {
+		// 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 {
+			return
+		}
+		r, err := d.regConn.Resolve(nodeID, d.NodeID())
+		if err != nil {
+			return
+		}
+		d.cacheResolve(nodeID, r)
+		resp = r
+	}
+	if relayOnly, _ := resp["relay_only"].(bool); relayOnly {
+		return
+	}
+	realAddrStr, _ := resp["real_addr"].(string)
+	if realAddrStr == "" {
+		return
+	}
+	realAddr, err := net.ResolveUDPAddr("udp", realAddrStr)
+	if err != nil {
+		return
+	}
+
+	// A prior blackhole flip may have PINNED the peer to relay; ClearRelayOnDirect
+	// will not promote a pinned peer. Unpin it — we've confirmed above it is not
+	// registry relay-only, so direct is allowed to win on its merits.
+	if d.tunnels.IsRelayPinned(nodeID) {
+		d.tunnels.SetRelayPeerPinned(nodeID, false)
+	}
+
+	// Coordinate the punch (opens the conntrack pinhole on both NATs).
+	d.tunnels.RequestHolePunch(nodeID)
+
+	probe := &protocol.Packet{
+		Version:  protocol.Version,
+		Flags:    protocol.FlagACK,
+		Protocol: protocol.ProtoControl,
+		Src:      d.Addr(),
+		Dst:      protocol.Addr{Network: 0, Node: nodeID},
+		SrcPort:  protocol.PortPing,
+		DstPort:  protocol.PortPing,
+		Seq:      1,
+	}
+	// Give the punch a moment to land, then probe the real address a few
+	// times so at least DirectClearsRequired (=3) direct decrypts arrive
+	// while the pinhole is open.
+	for i := 0; i < 5; i++ {
+		time.Sleep(200 * time.Millisecond)
+		if err := d.tunnels.SendDirectProbeTo(nodeID, realAddr, probe); err != nil {
+			slog.Debug("direct upgrade probe skipped", "node_id", nodeID, "error", err)
+			return
+		}
+	}
+}
+
 // ---------------------------------------------------------------------------
 // Network membership reconciliation: periodic delta detection vs the
 // registry's authoritative membership list. The reconciler is L7-only: