fix(pilotctl,install,daemon-tests): rc5 UX gaps + flaky panic-survival tests

Three UX gaps from the rc4 clean-install flow:

1. macOS install.sh installs a launchd plist (KeepAlive=true), so the
   daemon lifecycle is owned by launchd rather than by the PID file
   pilotctl writes when it forks. `pilotctl daemon stop` returned
   "daemon socket is active but PID file is missing" and refused to
   act; SIGTERM would have respawned anyway via KeepAlive.

   pilotctl now detects `~/Library/LaunchAgents/com.vulturelabs.pilot
   -daemon.plist` and routes start/stop through `launchctl bootstrap
   gui/<UID> <plist>` / `launchctl bootout gui/<UID>/<label>`. The
   pilotctl-fork path stays as the fallback for Linux + manual setups.

2. `pilotctl send-message <peer> --data ... --wait` would ACK the
   outbound but the responder's reply SYN was rejected by the local
   trust gate (`SYN rejected: untrusted source`). --wait then timed out
   with no actionable signal. The compiled-in trustedagents allowlist
   protects this for the agents it covers, but stale entries (a public
   agent redeployed under a new node ID) and any public peer not in the
   list fall through to the "pub=true → return silently" path.

   maybeAutoHandshake's Branch 3 (public peer, not in compiled list)
   now runs a best-effort handshake + WaitForTrust(5s) before
   returning. Public auto-approve peers finalise mutual trust in
   ~700-2400 ms; replies pass the local trust gate naturally afterward.

3. install.sh `PILOT_RC=1` hits api.github.com/repos/.../releases. When
   the unauthenticated 60/hr limit is exhausted the API returns 403,
   the existing code captured TAG="" silently, and the script fell
   through to source-build of `main` HEAD with no -ldflags version
   stamp — the user thought they got the latest RC but got an
   unstamped "dev" binary.

   install.sh now:
     - accepts `PILOT_RELEASE_TAG=vX.Y.Z-rcN` as an explicit override
       (skips the API entirely);
     - separates the API body from its HTTP status code so a 403 is
       detected explicitly + reported, with a clear suggestion to set
       PILOT_RELEASE_TAG or retry in ~1h, rather than silently
       degrading to a source build.

Also: pkg/daemon/zz_panic_survival_test.go — the 10 TestL*PanicSurvival
tests all called t.Parallel() while sharing the package-global
recoveredPanicCount counter. resetRecoveredPanicCountForTest() in one
test could wipe the increment of another's recoverLayer mid-assertion,
producing a flaky FAIL ("did not catch nil-packet panic") even though
the panic WAS recovered (CI logs showed recovered_total=1 immediately
before the FAIL). Drop t.Parallel() from all ten; each test is <50ms
so serialisation has no measurable cost.

Verified live on a clean install:
  - daemon start/stop round-trip via launchctl, no PID-file error;
  - untrust 179172 + send-message list-agents --wait now succeeds
    end-to-end with the reply (~2.2 KB directory dump) inline;
  - `go test -count=10 -run TestL.*PanicSurvival ./pkg/daemon/` passes.

Author: teovl

cmd/pilotctl/main.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -541,7 +542,28 @@ func maybeAutoHandshake(d *driver.Driver, addr protocol.Addr, skip bool) {
 			fmt.Sprintf("run: pilotctl handshake %s", addr),
 			"refusing tunnel to private node %s without trust", addr)
 	}
-	// Public peer, no trust needed — proceed.
+	// Public peer — best-effort handshake so replies survive our local
+	// trust gate. Without this, a request/reply pattern (e.g. send-message
+	// --wait) gets its ACK back but the responder's reply SYN is silently
+	// rejected by our daemon (daemon.go SYN-trust-gate). The peer is
+	// public and the embedded trustedagents allowlist may not cover every
+	// public hostname (entries can become stale when a public agent is
+	// re-deployed under a new node ID), so we treat any registry-resolved
+	// public peer the same as a Trusted Agent. Auto-approve peers (most
+	// public agents are) finalise mutual trust in ~700-2400 ms; non-auto-
+	// approve peers still see the request in pending and the tunnel
+	// proceeds best-effort.
+	if _, err := d.Handshake(addr.Node, "auto-handshake: public peer"); err != nil {
+		// Handshake send failure is non-fatal — fall through to send.
+		return
+	}
+	if resp, err := d.WaitForTrust(addr.Node, 5000); err == nil {
+		if trusted, _ := resp["trusted"].(bool); trusted {
+			if !jsonOutput {
+				fmt.Fprintf(os.Stderr, "auto-handshake established trust with public peer %s\n", addr)
+			}
+		}
+	}
 }
 
 func parseAddrOrHostname(d *driver.Driver, arg string) (protocol.Addr, error) {
@@ -2111,9 +2133,94 @@ func buildDaemonArgs(args []string) (daemonArgs []string, socketPath string) {
 	return daemonArgs, socketPath
 }
 
+// launchdAgentLabels enumerates known launchd labels for the daemon.
+// install.sh renamed `com.vulturelabs.pilot-daemon` to
+// `network.pilotprotocol.pilot-daemon`; the uninstall path handles both
+// for backward compat, and so does pilotctl. The first match wins.
+var launchdAgentLabels = []string{
+	"network.pilotprotocol.pilot-daemon",
+	"com.vulturelabs.pilot-daemon",
+}
+
+// launchdAgentPlist returns the path to the per-user launchd plist that
+// install.sh creates on macOS, if it exists; otherwise "". When non-empty
+// the daemon's lifecycle is owned by launchd (not by the PID file pilotctl
+// writes when it forks the daemon itself), and stop/start must go through
+// launchctl so KeepAlive doesn't immediately respawn what we kill.
+// Returns the plist path AND the label embedded in it.
+func launchdAgentPlist() (path, label string) {
+	if runtime.GOOS != "darwin" {
+		return "", ""
+	}
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", ""
+	}
+	for _, lbl := range launchdAgentLabels {
+		p := filepath.Join(home, "Library", "LaunchAgents", lbl+".plist")
+		if _, err := os.Stat(p); err == nil {
+			return p, lbl
+		}
+	}
+	return "", ""
+}
+
+// launchdAgentDomainTarget returns the gui/<UID>/<label> target used by
+// `launchctl bootout` / `launchctl kickstart`.
+func launchdAgentDomainTarget(label string) string {
+	return fmt.Sprintf("gui/%d/%s", os.Getuid(), label)
+}
+
+// launchdAgentLoaded reports whether the launchd label is currently
+// loaded in the user's gui domain (i.e. plist registered, KeepAlive etc.
+// are active even if the underlying process is briefly between restarts).
+func launchdAgentLoaded(label string) bool {
+	if label == "" {
+		return false
+	}
+	out, err := exec.Command("launchctl", "list", label).Output()
+	if err != nil {
+		return false
+	}
+	return len(out) > 0
+}
+
 func cmdDaemonStart(args []string) {
 	flags, _ := parseFlags(args)
 
+	// macOS install.sh installs a launchd plist. When present, route start
+	// through launchctl so the agent is registered and KeepAlive supervises
+	// the process; otherwise `pilotctl daemon stop` would have nothing to
+	// stop (KeepAlive immediately respawns) and the user sees flapping.
+	if plist, label := launchdAgentPlist(); plist != "" {
+		if launchdAgentLoaded(label) {
+			fatalHint("already_exists",
+				"stop it first with: pilotctl daemon stop",
+				"daemon is already running (launchd agent %s loaded)", label)
+		}
+		if err := exec.Command("launchctl", "bootstrap", fmt.Sprintf("gui/%d", os.Getuid()), plist).Run(); err != nil {
+			fatalCode("internal", "launchctl bootstrap: %v", err)
+		}
+		// Poll socket until daemon is responsive.
+		waitDeadline := time.Now().Add(10 * time.Second)
+		for time.Now().Before(waitDeadline) {
+			if d, err := driver.Connect(getSocket()); err == nil {
+				if _, err := d.Info(); err == nil {
+					d.Close()
+					if jsonOutput {
+						outputOK(map[string]interface{}{"managed_by": "launchd", "label": label})
+					} else {
+						fmt.Printf("daemon started via launchd (label %s)\n  Socket: %s\n", label, getSocket())
+					}
+					return
+				}
+				d.Close()
+			}
+			time.Sleep(200 * time.Millisecond)
+		}
+		fatalCode("timeout", "launchd loaded the agent but the socket did not become ready within 10s")
+	}
+
 	// Check if already running
 	if pid := readPID(); pid > 0 {
 		if processExists(pid) {
@@ -2254,6 +2361,32 @@ func cmdDaemonStart(args []string) {
 }
 
 func cmdDaemonStop() {
+	// macOS install.sh installs a launchd plist with KeepAlive=true. When
+	// we own it via launchd, SIGTERM to the PID respawns the process
+	// immediately — the right way to stop is `launchctl bootout`. Detect
+	// this case before falling back to PID-file management.
+	if plist, label := launchdAgentPlist(); plist != "" && launchdAgentLoaded(label) {
+		if err := exec.Command("launchctl", "bootout", launchdAgentDomainTarget(label)).Run(); err != nil {
+			fatalCode("internal", "launchctl bootout: %v", err)
+		}
+		// Wait for the daemon socket to disappear (launchd reaps the
+		// process); the agent stays unloaded across reboots until the
+		// user calls `pilotctl daemon start` (bootstrap) again.
+		waitDeadline := time.Now().Add(10 * time.Second)
+		for time.Now().Before(waitDeadline) {
+			if !launchdAgentLoaded(label) {
+				if jsonOutput {
+					outputOK(map[string]interface{}{"managed_by": "launchd", "label": label})
+				} else {
+					fmt.Printf("daemon stopped (launchd agent %s unloaded)\n", label)
+				}
+				return
+			}
+			time.Sleep(200 * time.Millisecond)
+		}
+		fatalCode("timeout", "launchctl bootout returned but agent did not unload within 10s")
+	}
+
 	pid := readPID()
 	if pid <= 0 {
 		// Try socket

pkg/daemon/zz_panic_survival_test.go

@@ -18,7 +18,6 @@ import (
 // length-check error paths; the boundary is defense-in-depth for any
 // future code change that introduces a slice OOB.
 func TestL1UnmarshalPanicSurvival(t *testing.T) {
-	t.Parallel()
 	cases := [][]byte{
 		nil,              // nil slice
 		{},               // empty
@@ -38,7 +37,6 @@ func TestL1UnmarshalPanicSurvival(t *testing.T) {
 // known panic-free path returns "payload too large" for oversized
 // payloads; the boundary protects against future panic introductions.
 func TestL1MarshalPanicSurvival(t *testing.T) {
-	t.Parallel()
 	huge := make([]byte, 0xFFFF+1)
 	pkt := &protocol.Packet{
 		Version:  protocol.Version,
@@ -60,7 +58,6 @@ func TestL1MarshalPanicSurvival(t *testing.T) {
 // burst. Real edge case: junk magic bytes, truncated headers, a too-
 // short PILS frame.
 func TestL2ReadLoopPanicSurvival(t *testing.T) {
-	t.Parallel()
 	tm := NewTunnelManager()
 	if err := tm.Listen("127.0.0.1:0"); err != nil {
 		t.Fatalf("Listen: %v", err)
@@ -126,7 +123,6 @@ func TestL2ReadLoopPanicSurvival(t *testing.T) {
 // deliver). The boundary keeps the function alive — process must not
 // crash. Real edge cases used.
 func TestL4BeaconMessagePanicSurvival(t *testing.T) {
-	t.Parallel()
 	tm := NewTunnelManager()
 	defer tm.Close()
 	from := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1}
@@ -151,7 +147,6 @@ func TestL4BeaconMessagePanicSurvival(t *testing.T) {
 // on a synthetic Daemon. With nil registry conn the function returns
 // cleanly; the boundary protects against future panic introductions.
 func TestL4BeaconRefreshTickPanicSurvival(t *testing.T) {
-	t.Parallel()
 	d := &Daemon{}
 	d.beaconRefreshTick(true)
 	d.beaconRefreshTick(false)
@@ -170,7 +165,6 @@ func TestL4BeaconRefreshTickPanicSurvival(t *testing.T) {
 // TestL5HandleAuthKeyExchangePanicSurvival drives handleAuthKeyExchange
 // with malformed inputs (too-short, all-zero). Real edge cases.
 func TestL5HandleAuthKeyExchangePanicSurvival(t *testing.T) {
-	t.Parallel()
 	tm := NewTunnelManager()
 	if err := tm.EnableEncryption(); err != nil {
 		t.Fatalf("EnableEncryption: %v", err)
@@ -184,7 +178,6 @@ func TestL5HandleAuthKeyExchangePanicSurvival(t *testing.T) {
 // TestL5HandleKeyExchangePanicSurvival drives handleKeyExchange with
 // malformed inputs.
 func TestL5HandleKeyExchangePanicSurvival(t *testing.T) {
-	t.Parallel()
 	tm := NewTunnelManager()
 	if err := tm.EnableEncryption(); err != nil {
 		t.Fatalf("EnableEncryption: %v", err)
@@ -200,7 +193,6 @@ func TestL5HandleKeyExchangePanicSurvival(t *testing.T) {
 // triggers a rate-limited rekey request and returns. The boundary
 // keeps the daemon alive across a burst.
 func TestL6HandleEncryptedPanicSurvival(t *testing.T) {
-	t.Parallel()
 	tm := NewTunnelManager()
 	if err := tm.EnableEncryption(); err != nil {
 		t.Fatalf("EnableEncryption: %v", err)
@@ -226,7 +218,6 @@ func TestL6HandleEncryptedPanicSurvival(t *testing.T) {
 // Uses test-only injection: replacing RetxSend with a panicking
 // closure is the cleanest way to hit a real code path with a panic.
 func TestL7RetxLoopPanicSurvival(t *testing.T) {
-	t.Parallel()
 	d := New(Config{
 		ListenAddr:   "127.0.0.1:0",
 		IdentityPath: t.TempDir() + "/identity.json",
@@ -285,7 +276,6 @@ func TestL7RetxLoopPanicSurvival(t *testing.T) {
 // a nil packet, which would nil-deref on pkt.Protocol. The boundary
 // must catch it.
 func TestL7RoutePacketPanicSurvival(t *testing.T) {
-	t.Parallel()
 	d := New(Config{
 		ListenAddr:   "127.0.0.1:0",
 		IdentityPath: t.TempDir() + "/identity.json",
@@ -310,7 +300,6 @@ func TestL7RoutePacketPanicSurvival(t *testing.T) {
 // no real unrecovered panic surface exists today. This test guarantees
 // the L9 boundary mechanism would catch one if introduced.
 func TestL9IPCHandlerPanicSurvival(t *testing.T) {
-	t.Parallel()
 	d := New(Config{
 		ListenAddr:   "127.0.0.1:0",
 		IdentityPath: t.TempDir() + "/identity.json",

tests/zz_eventstream_broker_parity_test.go

@@ -136,29 +136,49 @@ func TestBrokerParityNormal(t *testing.T) {
 	}
 
 	// Receive must preserve topic + payload bytes exactly.
+	// waitForSubscription's probes ride the same topic, so the broker may
+	// deliver a left-over probe before the real event arrives. Loop on Recv
+	// and skip probes; bail only on the real payload (or timeout). The
+	// previous single-Recv form was flaky on macOS CI where the probe
+	// reached the subscriber before the real publish.
 	deadline := time.After(5 * time.Second)
 	type res struct {
 		evt *eventstream.Event
 		err error
 	}
-	ch := make(chan res, 1)
+	stopRecv := make(chan struct{})
+	recvCh := make(chan res, 4)
 	go func() {
-		evt, err := sub.Recv()
-		ch <- res{evt, err}
+		for {
+			evt, err := sub.Recv()
+			select {
+			case recvCh <- res{evt, err}:
+			case <-stopRecv:
+				return
+			}
+			if err != nil {
+				return
+			}
+		}
 	}()
+	defer close(stopRecv)
 
 	// Re-publish on a ticker in case the probe drained the real one.
 	tick := time.NewTicker(100 * time.Millisecond)
 	defer tick.Stop()
 	for {
 		select {
-		case r := <-ch:
+		case r := <-recvCh:
 			if r.err != nil {
 				t.Fatalf("recv: %v", r.err)
 			}
 			if r.evt.Topic != "parity.normal" {
 				t.Fatalf("topic = %q, want %q", r.evt.Topic, "parity.normal")
 			}
+			if string(r.evt.Payload) == "__probe__" {
+				// Subscription-readiness probe — skip and keep reading.
+				continue
+			}
 			if string(r.evt.Payload) != string(want) {
 				t.Fatalf("payload mismatch: got %q want %q", r.evt.Payload, want)
 			}

web/public/install.sh

@@ -194,12 +194,33 @@ trap 'rm -rf "$TMPDIR"' EXIT
 
 ARCHIVE="pilot-${OS}-${ARCH}.tar.gz"
 
-# Resolve the latest release tag.
-# - Default path uses the unauthenticated /releases/latest/download/ redirect,
-#   which is not subject to the 60/hr api.github.com rate limit.
-# - PILOT_RC=1 still hits the API because pre-releases need the listing endpoint.
-if [ "${PILOT_RC:-}" = "1" ]; then
-    TAG=$(curl -fsSL "https://api.github.com/repos/${REPO}/releases" 2>/dev/null | grep '"tag_name"' | head -1 | cut -d'"' -f4 || true)
+# Resolve the release tag.
+# - PILOT_RELEASE_TAG=v1.2.3  : explicit override, no network round-trip.
+# - PILOT_RC=1                : list releases via api.github.com and pick the
+#                               newest (incl. pre-releases). 403 rate-limited
+#                               output is detected and reported instead of
+#                               silently falling through to source-build with
+#                               an unstamped version.
+# - default                   : follow the /releases/latest/download/ redirect,
+#                               which uses the unauthenticated CDN and is not
+#                               subject to the 60/hr api.github.com rate limit.
+if [ -n "${PILOT_RELEASE_TAG:-}" ]; then
+    TAG="$PILOT_RELEASE_TAG"
+elif [ "${PILOT_RC:-}" = "1" ]; then
+    API_BODY="$TMPDIR/releases.json"
+    API_CODE=$(curl -sSL -o "$API_BODY" -w '%{http_code}' "https://api.github.com/repos/${REPO}/releases" 2>/dev/null || echo "000")
+    if [ "$API_CODE" = "403" ]; then
+        echo "Error: GitHub API rate-limited (403) while resolving the latest pre-release." >&2
+        echo "  Workarounds:" >&2
+        echo "    - retry in ~1 hour, OR" >&2
+        echo "    - pin the tag:  PILOT_RELEASE_TAG=vX.Y.Z-rcN curl ... | sh" >&2
+        echo "  Refusing to silently source-build an unstamped binary." >&2
+        exit 1
+    fi
+    if [ "$API_CODE" = "200" ]; then
+        TAG=$(grep '"tag_name"' "$API_BODY" | head -1 | cut -d'"' -f4 || true)
+    fi
+    rm -f "$API_BODY"
 else
     TAG=$(curl -fsSI "https://github.com/${REPO}/releases/latest/download/${ARCHIVE}" 2>/dev/null \
         | grep -i '^location:' \