macOS: repair network recovery and improve diagnostics

cmd/cli/dns_intercept_darwin.go

@@ -7,6 +7,7 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"net"
+	"net/netip"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -75,6 +76,11 @@ const (
 
 const (
 	// pfProbeDomain is the suffix used for pf interception probe queries.
+	// The full probe domain is "_pf-probe-<hex>.<pfProbeDomain>".
+	// These queries are sent by a subprocess WITHOUT the _ctrld group GID,
+	// so pf should intercept them and redirect to ctrld. If ctrld receives
+	// the query, interception is working. A sent query without receipt permits
+	// repair. An unsent or unavailable probe is indeterminate.
 	// No trailing dot — canonicalName() in the DNS handler strips trailing dots.
 	pfProbeDomain = "pf-probe.ctrld.test"
 
@@ -1505,10 +1511,19 @@ func (p *prog) finishPFStabilization(stableRequired time.Duration) {
 }
 
 // probePFInterceptFn and forceReloadPFInterceptFn are the functional verification seams
-// shared by both probers.
+// shared by all four probe callers.
 var (
 	probePFInterceptFn       = (*prog).probePFIntercept
 	forceReloadPFInterceptFn = (*prog).forceReloadPFMainRuleset
+	pfProbeNameservers       = ctrld.OsResolverNameservers
+	newPFProbeCommand        = func(ctx context.Context, host, packet string) *exec.Cmd {
+		cmd := exec.CommandContext(ctx, os.Args[0], "pf-probe-send", host, packet)
+		// wheel, not _ctrld: the probe must traverse interception.
+		cmd.SysProcAttr = &syscall.SysProcAttr{
+			Credential: &syscall.Credential{Uid: 0, Gid: 0},
+		}
+		return cmd
+	}
 )
 
 // pfFunctionalProbeOwnerWait bounds how long the post-stabilization verifier waits for
@@ -1579,17 +1594,29 @@ func (p *prog) verifyInterceptAfterStabilization() {
 		return
 	}
 
-	if probePFInterceptFn(p) {
+	result := probePFInterceptFn(p)
+	if result.result == pfProbeIndeterminate {
+		mainLog.Load().Debug().Msg("DNS intercept: post-stabilization probe indeterminate; repair deferred")
+		return
+	}
+	if result.result == pfProbeIntercepted {
 		mainLog.Load().Debug().Msg("DNS intercept: post-stabilization probe passed — interception is translating")
 		return
 	}
 
 	mainLog.Load().Warn().Msg("DNS intercept: post-stabilization rules are intact but the probe FAILED — forcing one reload")
-	if !forceReloadPFInterceptFn(p) {
+	if !p.repairPFIntercept(result, "post_stabilization") {
 		mainLog.Load().Error().Msg("DNS intercept: post-stabilization forced reload did not run — leaving recovery to the watchdog")
 		return
 	}
-	if probePFInterceptFn(p) {
+	cause := result
+	result = probePFInterceptFn(p)
+	logPFRepair(cause, "post_stabilization", result.result.String(), true, result)
+	if result.result == pfProbeIndeterminate {
+		mainLog.Load().Warn().Msg("DNS intercept: post-stabilization reload completed; verification indeterminate")
+		return
+	}
+	if result.result == pfProbeIntercepted {
 		mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-stabilization reload")
 		return
 	}
@@ -1628,8 +1655,8 @@ var pfWakeProbeOwnerWait = 250 * time.Millisecond
 // drive the post-stabilization verifier; without that reconnect nothing would have probed.
 //
 // The OS resolver list is refreshed before every probe. probePFIntercept aims at the
-// first OS nameserver and reports success when it has no target, so a list still holding
-// a pre-sleep VPN resolver turns this into a meaningless pass or a false failure.
+// first usable IPv4 LAN target, or the original usable first public target.
+// No target is indeterminate. Refreshing avoids probing a stale VPN resolver.
 func (p *prog) verifyInterceptAfterWake(gap time.Duration) {
 	if p.dnsInterceptState == nil {
 		return
@@ -1681,7 +1708,7 @@ func (p *prog) verifyInterceptAfterWake(gap time.Duration) {
 		mainLog.Load().Warn().Msg("DNS intercept: post-wake verification never probed - every attempt was deferred or skipped - the watchdog will retry")
 		return
 	}
-	mainLog.Load().Error().Msgf("DNS intercept: interception still not translating after resume (%d probe attempts, %d forced reloads) - the watchdog will retry",
+	mainLog.Load().Error().Msgf("DNS intercept: interception not verified after resume (%d probe attempts, %d forced reloads) - the watchdog will retry",
 		probed, repairs)
 }
 
@@ -1708,7 +1735,12 @@ func (p *prog) waitBeforeWakeProbe(delay time.Duration) bool {
 // The caller holds functional-probe ownership for the call.
 func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired bool) {
 	initializeOsResolver(ctrld.LoggerCtx(context.Background(), mainLog.Load()), true)
-	if probePFInterceptFn(p) {
+	result := probePFInterceptFn(p)
+	if result.result == pfProbeIndeterminate {
+		mainLog.Load().Debug().Msg("DNS intercept: post-wake probe indeterminate; repair deferred")
+		return false, false
+	}
+	if result.result == pfProbeIntercepted {
 		if attempt > 0 || repairs > 0 {
 			mainLog.Load().Info().Msgf("DNS intercept: post-wake probe %d/%d passed - interception is translating", attempt+1, len(pfWakeProbeDelays))
 		} else {
@@ -1720,16 +1752,19 @@ func (p *prog) runWakeProbeAttempt(attempt, repairs int) (restored, repaired boo
 	if repairs >= pfWakeMaxRepairs {
 		mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - repair budget of %d forced reloads is spent, not reloading again",
 			attempt+1, len(pfWakeProbeDelays), pfWakeMaxRepairs)
+		logPFRepair(result, "post_wake", "budget_exhausted", false, pfProbeObservation{})
 		return false, false
 	}
 
 	mainLog.Load().Warn().Msgf("DNS intercept: post-wake probe %d/%d FAILED - rules survived the suspend but pf is not translating, forcing a reload",
 		attempt+1, len(pfWakeProbeDelays))
-	if !forceReloadPFInterceptFn(p) {
+	if !p.repairPFIntercept(result, "post_wake") {
 		mainLog.Load().Error().Msg("DNS intercept: post-wake forced reload did not run")
 		return false, false
 	}
-	if probePFInterceptFn(p) {
+	confirmation := probePFInterceptFn(p)
+	logPFRepair(result, "post_wake", confirmation.result.String(), true, confirmation)
+	if confirmation.result == pfProbeIntercepted {
 		mainLog.Load().Info().Msg("DNS intercept: interception restored by the post-wake reload")
 		return true, true
 	}
@@ -2042,18 +2077,7 @@ func (p *prog) pfWatchdog() {
 			ensureInterceptDNSTargetFn(p, []string{})
 
 			result := p.ensurePFAnchorActive()
-			if result == pfAnchorCheckIntact {
-				// Only an authoritative intact result may trigger the functional probe.
-				// Skipped/backoff/failed checks must not be treated as healthy text state.
-				if !p.pfMonitorRunning.Load() && !p.probePFIntercept() {
-					mainLog.Load().Warn().Msg("DNS intercept watchdog: rules intact but probe FAILED — forcing full reload")
-					if p.forceReloadPFMainRuleset() {
-						result = pfAnchorCheckRestored
-					} else {
-						result = pfAnchorCheckFailed
-					}
-				}
-			}
+			result = p.checkPFWatchdogProbe(result)
 
 			if result == pfAnchorCheckIntact {
 				// Check if backoff should be reset only after an authoritative healthy check.
@@ -2108,6 +2132,27 @@ func (p *prog) pfWatchdog() {
 	}
 }
 
+// A successful forced reload following a probe miss is not evidence that an
+// anchor was missing. Keep that repair out of the missing-anchor counter.
+func (p *prog) checkPFWatchdogProbe(result pfAnchorCheckResult) pfAnchorCheckResult {
+	if result != pfAnchorCheckIntact || p.pfMonitorRunning.Load() {
+		return result
+	}
+	probe := probePFInterceptFn(p)
+	switch probe.result {
+	case pfProbeIntercepted:
+		return result
+	case pfProbeNotIntercepted:
+		mainLog.Load().Warn().Msg("DNS intercept watchdog: sent probe was not intercepted; forcing full reload")
+		if !p.repairPFIntercept(probe, "watchdog") {
+			mainLog.Load().Warn().Msg("DNS intercept watchdog: probe repair did not complete; will retry")
+		}
+	case pfProbeIndeterminate:
+		mainLog.Load().Debug().Msg("DNS intercept watchdog: probe indeterminate; repair deferred")
+	}
+	return pfAnchorCheckDeferred
+}
+
 // exemptVPNDNSServers updates the pf anchor rules with interface-scoped exemptions
 // for VPN DNS servers, allowing VPN local DNS handlers (e.g., Tailscale MagicDNS
 // via Network Extension) to receive DNS queries from all processes on their interface.
@@ -2175,67 +2220,91 @@ func (p *prog) exemptVPNDNSServers(exemptions []vpnDNSExemption) error {
 	return nil
 }
 
-// probePFIntercept tests whether pf's rdr translation is actually working by
-// sending a DNS query through the interception path from a subprocess that does
-// NOT have the _ctrld group GID. If pf interception is working, the query gets
-// redirected to 127.0.0.1:53 (ctrld), and the DNS handler signals us. If broken
-// (rdr rules present but not evaluating), the query goes to the real DNS server
-// and we time out.
-//
-// Returns true if interception is working, false if broken or indeterminate.
-func (p *prog) probePFIntercept() bool {
+// Keep warning coalescing available throughout intercept startup and recovery.
+var pfProbeLogs pfProbeLogState
+
+// probePFIntercept distinguishes a meaningful interception failure from a probe
+// that could not send. Only pfProbeNotIntercepted permits destructive repair.
+func (p *prog) probePFIntercept() pfProbeObservation {
+	observed := pfProbeObservation{
+		probeID:            fmt.Sprintf("_pf-probe-%x.%s", time.Now().UnixNano(), pfProbeDomain),
+		recoveryGeneration: p.recoveryGen.Load(),
+		stage:              "target", code: "unavailable",
+	}
 	if p.dnsInterceptState == nil {
-		return true
+		return observed
 	}
-
-	nsIPs := ctrld.OsResolverNameservers()
-	if len(nsIPs) == 0 {
-		mainLog.Load().Debug().Msg("DNS intercept probe: no OS resolver nameservers available")
-		return true // can't probe without a target
+	started := time.Now()
+	servers := pfProbeNameservers()
+	host := pfProbeTarget(servers)
+	family := "none"
+	if len(servers) > 0 {
+		if first, _, err := net.SplitHostPort(servers[0]); err == nil {
+			if addr, err := netip.ParseAddr(first); err == nil {
+				observed.target = net.JoinHostPort(addr.String(), "53")
+				family = "ipv6"
+				if addr.Is4() {
+					family = "ipv4"
+				}
+			}
+		}
 	}
-	host, _, _ := net.SplitHostPort(nsIPs[0])
-	if host == "" || host == "127.0.0.1" || host == "::1" {
-		mainLog.Load().Debug().Msg("DNS intercept probe: OS resolver is localhost, skipping probe")
-		return true // can't probe through localhost
+	if host != "" {
+		observed.target = net.JoinHostPort(host, "53")
+		family = "ipv4"
+		probeCh, deregister := p.registerInterceptProbe(observed.probeID)
+		defer deregister()
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		go func() {
+			select {
+			case <-p.stopCh:
+				cancel()
+			case <-ctx.Done():
+			}
+		}()
+		cmd := newPFProbeCommand(ctx, host, fmt.Sprintf("%x", buildDNSQueryPacket(observed.probeID)))
+		result := runPFProbe(ctx, cmd, probeCh, pfProbeTimeout)
+		observed.result, observed.stage, observed.code = result.result, result.stage, result.code
 	}
-
-	// Generate unique probe domain
-	probeID := fmt.Sprintf("_pf-probe-%x.%s", time.Now().UnixNano()&0xFFFFFFFF, pfProbeDomain)
-
-	// Register this attempt's own domain: overlapping probes must not cancel each other.
-	probeCh, deregister := p.registerInterceptProbe(probeID)
-	defer deregister()
-
-	// Build a minimal DNS query packet for the probe domain.
-	// We use exec.Command to send from a subprocess with GID=0 (wheel),
-	// so pf's _ctrld group exemption does NOT apply and the query gets intercepted.
-	dnsPacket := buildDNSQueryPacket(probeID)
-
-	// Send via a helper subprocess that drops the _ctrld group
-	cmd := exec.Command(os.Args[0], "pf-probe-send", host, fmt.Sprintf("%x", dnsPacket))
-	cmd.SysProcAttr = &syscall.SysProcAttr{
-		Credential: &syscall.Credential{
-			Uid: 0,
-			Gid: 0, // wheel group — NOT _ctrld, so pf intercepts it
-		},
+	// Cancellation is a shutdown diagnostic, not a change in PF health.
+	retain, repeats := false, uint64(0)
+	if observed.code != "canceled" {
+		retain, repeats = pfProbeLogs.change(observed)
 	}
-
-	if err := cmd.Start(); err != nil {
-		mainLog.Load().Debug().Err(err).Msg("DNS intercept probe: failed to start probe subprocess")
-		return true // can't probe, assume OK
+	event := mainLog.Load().Debug()
+	if retain {
+		event = mainLog.Load().Warn()
 	}
+	event.Str("probe_id", observed.probeID).Uint64("recovery_generation", observed.recoveryGeneration).
+		Str("resolver_target", observed.target).Str("resolver_family", family).
+		Str("stage", observed.stage).Str("error_code", observed.code).
+		Str("outcome", observed.result.String()).Uint64("repeated_results", repeats).
+		Bool("repair_eligible", observed.result == pfProbeNotIntercepted).
+		Int64("elapsed_ms", time.Since(started).Milliseconds()).Msg("DNS intercept probe result")
+	return observed
+}
 
-	// Don't leak the subprocess
-	go func() {
-		_ = cmd.Wait()
-	}()
-
-	select {
-	case <-probeCh:
-		return true
-	case <-time.After(pfProbeTimeout):
-		return false
+// Carry the cause by value. A concurrent probe cannot replace its correlation fields.
+func (p *prog) repairPFIntercept(cause pfProbeObservation, caller string) bool {
+	logPFRepair(cause, caller, "started", false, pfProbeObservation{})
+	performed := forceReloadPFInterceptFn(p)
+	outcome := "not_run"
+	if performed {
+		outcome = "reload_completed"
 	}
+	logPFRepair(cause, caller, outcome, performed, pfProbeObservation{})
+	return performed
+}
+
+func logPFRepair(cause pfProbeObservation, caller, outcome string, performed bool, confirmation pfProbeObservation) {
+	mainLog.Load().Warn().Str("probe_id", cause.probeID).
+		Uint64("recovery_generation", cause.recoveryGeneration).
+		Str("caller", caller).Str("cause", cause.result.String()).
+		Str("stage", cause.stage).Str("error_code", cause.code).
+		Str("resolver_target", cause.target).Str("outcome", outcome).
+		Bool("reload_performed", performed).Str("confirmation_probe_id", confirmation.probeID).
+		Msg("PF repair")
 }
 
 // buildDNSQueryPacket constructs a minimal DNS query packet (wire format) for the given domain.
@@ -2272,6 +2341,8 @@ func buildDNSQueryPacket(domain string) []byte {
 // block traffic until ctrld loads rules into it.
 func cleanupStaleDNSInterceptState() {}
 
+var pfInterceptMonitorDelays = []time.Duration{0, 500 * time.Millisecond, time.Second, 2 * time.Second, 4 * time.Second}
+
 // pfInterceptMonitor runs asynchronously after interface changes are detected.
 // It probes pf interception with exponential backoff and forces a full pf reload
 // if the probe fails. Only one instance runs at a time (singleton via atomic.Bool).
@@ -2296,7 +2367,7 @@ func (p *prog) pfInterceptMonitor() {
 
 	// Backoff schedule: probe quickly first, then space out.
 	// Total monitoring window: ~0 + 0.5 + 1 + 2 + 4 = ~7.5s
-	delays := []time.Duration{0, 500 * time.Millisecond, time.Second, 2 * time.Second, 4 * time.Second}
+	delays := pfInterceptMonitorDelays
 
 	for i, delay := range delays {
 		if delay > 0 {
@@ -2307,22 +2378,33 @@ func (p *prog) pfInterceptMonitor() {
 			return
 		}
 
-		if probePFInterceptFn(p) {
+		result := probePFInterceptFn(p)
+		if result.result == pfProbeIndeterminate {
+			mainLog.Load().Debug().Msg("DNS intercept monitor: probe indeterminate; repair deferred")
+			continue
+		}
+		if result.result == pfProbeIntercepted {
 			mainLog.Load().Debug().Msgf("DNS intercept monitor: probe %d/%d passed", i+1, len(delays))
 			continue // working now — keep monitoring in case it breaks later in the window
 		}
 
 		// Probe failed — pf translation is broken. Force full reload.
 		mainLog.Load().Warn().Msgf("DNS intercept monitor: probe %d/%d FAILED — pf translation broken, forcing full ruleset reload", i+1, len(delays))
-		forceReloadPFInterceptFn(p)
+		if !p.repairPFIntercept(result, "monitor") {
+			continue
+		}
 
 		// Verify the reload fixed it
 		time.Sleep(200 * time.Millisecond)
-		if probePFInterceptFn(p) {
-			mainLog.Load().Info().Msg("DNS intercept monitor: probe passed after reload — interception restored")
-			// Continue monitoring in case the hypervisor reloads pf again
-		} else {
-			mainLog.Load().Error().Msg("DNS intercept monitor: probe still failing after reload — pf may need manual intervention")
+		confirmation := probePFInterceptFn(p)
+		logPFRepair(result, "monitor", confirmation.result.String(), true, confirmation)
+		switch confirmation.result {
+		case pfProbeIntercepted:
+			mainLog.Load().Info().Msg("DNS intercept monitor: probe passed after reload - interception restored")
+		case pfProbeNotIntercepted:
+			mainLog.Load().Warn().Msg("DNS intercept monitor: probe still failing after reload - pf may need manual intervention")
+		case pfProbeIndeterminate:
+			mainLog.Load().Warn().Msg("DNS intercept monitor: reload completed; verification indeterminate")
 		}
 	}
 

cmd/cli/dns_intercept_darwin_test.go

@@ -838,13 +838,16 @@ func stubStabilizationProbe(t *testing.T, probeResults []bool, reloadOK bool) (p
 		probePFInterceptFn, forceReloadPFInterceptFn = originalProbe, originalReload
 	})
 	probeCalls, reloadCalls := 0, 0
-	probePFInterceptFn = func(*prog) bool {
+	probePFInterceptFn = func(*prog) pfProbeObservation {
 		result := false
 		if probeCalls < len(probeResults) {
 			result = probeResults[probeCalls]
 		}
 		probeCalls++
-		return result
+		if result {
+			return pfProbeObservation{result: pfProbeIntercepted}
+		}
+		return pfProbeObservation{result: pfProbeNotIntercepted}
 	}
 	forceReloadPFInterceptFn = func(*prog) bool {
 		reloadCalls++

cmd/cli/dns_proxy.go

@@ -1823,7 +1823,7 @@ func FlushDNSCache() error {
 
 // monitorNetworkChanges starts monitoring for network interface changes
 func (p *prog) monitorNetworkChanges(ctx context.Context) error {
-	mon, err := netmon.New(func(format string, args ...any) {
+	mon, err := newNetworkChangeMonitorFn(func(format string, args ...any) {
 		// Always fetch the latest logger (and inject the prefix)
 		p.logger.Load().Printf("netmon: "+format, args...)
 	})
@@ -1832,195 +1832,279 @@ func (p *prog) monitorNetworkChanges(ctx context.Context) error {
 	}
 
 	mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) {
-		isMajorChange := mon.IsMajorChangeFrom(delta.Old, delta.New)
-		p.handleDNS64NetworkChange(delta, isMajorChange)
-
-		// Get map of valid interfaces
-		validIfaces := ctrld.ValidInterfaces(ctrld.LoggerCtx(ctx, p.logger.Load()))
-
-		p.Debug().
-			Interface("old_state", delta.Old).
-			Interface("new_state", delta.New).
-			Bool("is_major_change", isMajorChange).
-			Msg("Network change detected")
-
-		changed := false
-		activeInterfaceExists := false
-		var changeIPs []netip.Prefix
-		// Check each valid interface for changes
-		for ifaceName := range validIfaces {
-			oldIface, oldExists := delta.Old.Interface[ifaceName]
-			newIface, newExists := delta.New.Interface[ifaceName]
-			if !newExists {
-				continue
-			}
-
-			oldIPs := delta.Old.InterfaceIPs[ifaceName]
-			newIPs := delta.New.InterfaceIPs[ifaceName]
-
-			// if a valid interface did not exist in old
-			// check that its up and has usable IPs
-			if !oldExists {
-				// The interface is new (was not present in the old state).
-				usableNewIPs := filterUsableIPs(newIPs)
-				if newIface.IsUp() && len(usableNewIPs) > 0 {
-					changed = true
-					changeIPs = usableNewIPs
-					p.Debug().
-						Str("interface", ifaceName).
-						Interface("new_ips", usableNewIPs).
-						Msg("Interface newly appeared (was not present in old state)")
-					break
-				}
-				continue
-			}
-
-			// Filter new IPs to only those that are usable.
-			usableNewIPs := filterUsableIPs(newIPs)
-
-			// Check if interface is up and has usable IPs.
-			if newIface.IsUp() && len(usableNewIPs) > 0 {
-				activeInterfaceExists = true
-			}
-
-			// Compare interface states and IPs (interfaceIPsEqual will itself filter the IPs).
-			if !interfaceStatesEqual(&oldIface, &newIface) || !interfaceIPsEqual(oldIPs, newIPs) {
-				if newIface.IsUp() && len(usableNewIPs) > 0 {
-					changed = true
-					changeIPs = usableNewIPs
-					p.Debug().
-						Str("interface", ifaceName).
-						Interface("old_ips", oldIPs).
-						Interface("new_ips", usableNewIPs).
-						Msg("Interface state or IPs changed")
-					break
-				}
-			}
-		}
-
-		// if the default route changed, set changed to true
-		if delta.New.DefaultRouteInterface != delta.Old.DefaultRouteInterface {
-			changed = true
-			p.Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface)
-		}
-
-		if !changed {
-			p.Debug().Msg("Ignoring interface change - no valid interfaces affected")
-			// check if the default IPs are still on an interface that is up
-			ValidateDefaultLocalIPsFromDelta(delta.New)
-			// Minor interface changes can still accompany pf/WFP or VPN DNS changes.
-			// On macOS, bound the immediate full reconciliation so link-local-only
-			// notification storms do not run pfctl/scutil work for every event.
-			// Windows keeps the existing immediate behavior. Tunnel changes always
-			// bypass the macOS limit, and delayed checks provide a trailing refresh.
-			if dnsIntercept && p.dnsInterceptState != nil {
-				p.handleDNSInterceptIgnoredNetworkChange(delta, time.Now())
-			}
-			return
-		}
-
-		if !activeInterfaceExists {
-			p.Debug().Msg("No active interfaces found, skipping reinitialization")
-			return
-		}
-
-		p.Debug().Msg("Link state changed, re-bootstrapping")
-		for _, uc := range p.cfg.Upstream {
-			uc.ReBootstrap(ctrld.LoggerCtx(ctx, p.logger.Load()))
-		}
-
-		// Get IPs from default route interface in new state
-		selfIP := p.defaultRouteIP()
-
-		// Ensure that selfIP is an IPv4 address.
-		// If defaultRouteIP mistakenly returns an IPv6 (such as a ULA), clear it
-		if ip := net.ParseIP(selfIP); ip != nil && ip.To4() == nil {
-			p.Debug().Msgf("DefaultRouteIP returned a non-ipv4 address: %s, ignoring it", selfIP)
-			selfIP = ""
-		}
-		var ipv6 string
-
-		if delta.New.DefaultRouteInterface != "" {
-			p.Debug().Msgf("Default route interface: %s, ips: %v", delta.New.DefaultRouteInterface, delta.New.InterfaceIPs[delta.New.DefaultRouteInterface])
-			for _, ip := range delta.New.InterfaceIPs[delta.New.DefaultRouteInterface] {
-				ipAddr, _ := netip.ParsePrefix(ip.String())
-				addr := ipAddr.Addr()
-				if selfIP == "" && addr.Is4() {
-					p.Debug().Msgf("Checking ip: %s", addr.String())
-					if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
-						selfIP = addr.String()
-					}
-				}
-				if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
-					ipv6 = addr.String()
-				}
-			}
-		} else {
-			// If no default route interface is set yet, use the changed IPs
-			p.Debug().Msgf("No default route interface found, using changed ips: %v", changeIPs)
-			for _, ip := range changeIPs {
-				ipAddr, _ := netip.ParsePrefix(ip.String())
-				addr := ipAddr.Addr()
-				if selfIP == "" && addr.Is4() {
-					p.Debug().Msgf("Checking ip: %s", addr.String())
-					if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
-						selfIP = addr.String()
-					}
-				}
-				if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
-					ipv6 = addr.String()
-				}
-			}
-		}
-
-		// Only set the IPv4 default if selfIP is a valid IPv4 address.
-		if ip := net.ParseIP(selfIP); ip != nil && ip.To4() != nil {
-			ctrld.SetDefaultLocalIPv4(ctrld.LoggerCtx(ctx, p.logger.Load()), ip)
-			if !isMobile() && p.ciTable != nil {
-				p.ciTable.SetSelfIP(selfIP)
-			}
-		}
-		if ip := net.ParseIP(ipv6); ip != nil {
-			ctrld.SetDefaultLocalIPv6(ctrld.LoggerCtx(ctx, p.logger.Load()), ip)
-		}
-		p.Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6)
-
-		p.debounceRecovery()
-
-		// Firewall mode: flush allowlist on network changes. Stale IPs from
-		// the old network may no longer be routable. DNS queries on the new
-		// network will repopulate the allowlist.
-		p.firewallOnNetworkChange()
-
-		// After network changes, verify our pf anchor is still active and
-		// refresh VPN DNS state. Order matters: tunnel checks first (may rebuild
-		// anchor), then VPN DNS refresh (updates exemptions in anchor), then
-		// delayed re-checks for async VPN teardown.
-		if dnsIntercept && p.dnsInterceptState != nil {
-			if !p.pfStabilizing.Load() {
-				p.ensurePFAnchorActive()
-			}
-			// Check tunnel interfaces unconditionally — it decides internally
-			// whether to enter stabilization or rebuild immediately.
-			p.checkTunnelInterfaceChanges()
-			// Refresh VPN DNS routes — runs after tunnel checks so the anchor
-			// rebuild includes current VPN DNS exemptions.
-			if p.vpnDNS != nil {
-				p.vpnDNS.Refresh(ctrld.LoggerCtx(ctx, p.logger.Load()), true)
-			}
-			// Rebuild the anchor if a VM/container bridge appeared/disappeared
-			// on this network change (no-op when the forwarded-source set is unchanged).
-			p.reconcileForwardedSources()
-			// Schedule delayed re-checks to catch async VPN teardown changes.
-			p.scheduleDelayedRechecks()
-		}
+		p.handleNetworkChange(ctx, delta, mon.IsMajorChangeFrom(delta.Old, delta.New))
 	})
 
 	mon.Start()
-	p.Debug().Msg("Network monitor started")
+	p.logger.Load().Debug().Msg("Network monitor started")
 	return nil
 }
 
+// handleNetworkChange is the network monitor callback, kept separate for synthetic deltas.
+func (p *prog) handleNetworkChange(ctx context.Context, delta *netmon.ChangeDelta, isMajorChange bool) {
+	// netmon v1.74.0 dispatches callbacks concurrently and caches only major
+	// snapshots. Minor events refer to that cache through Old. The synthetic
+	// delta.Major flag after a time jump does not change the cache.
+	p.networkSourceMu.Lock()
+	currentState := networkChangeCurrentStateFn(delta)
+	if !networkSnapshotCurrent(delta, isMajorChange, currentState) {
+		p.networkSourceMu.Unlock()
+		p.logger.Load().Debug().Msg("Network transition skipped: snapshot superseded")
+		return
+	}
+	transitionID := p.networkTransitionGen.Add(1)
+	sourceState := delta.New
+	// A late major callback and reordered minor callbacks share one cache
+	// epoch. For these cases, source validity comes from the OS, not ordering.
+	if delta.Monitor != nil && (!isMajorChange || (p.networkSourceEpoch == currentState && p.networkSourceState != currentState)) {
+		fresh, err := readNetworkSourceStateFn()
+		if err != nil || fresh == nil {
+			if !p.networkSourceReadFailed {
+				p.logger.Load().Warn().Uint64("transition_id", transitionID).
+					Msg("Network source snapshot unavailable; source writes deferred")
+			}
+			p.networkSourceReadFailed = true
+			sourceState = nil
+		} else {
+			snapshot := *fresh
+			snapshot.DefaultRouteInterface = currentState.DefaultRouteInterface
+			sourceState = &snapshot
+			if p.networkSourceReadFailed {
+				p.logger.Load().Warn().Uint64("transition_id", transitionID).Msg("Network source snapshot available again")
+			}
+			p.networkSourceReadFailed = false
+		}
+	}
+	p.networkSourceState, p.networkSourceEpoch = sourceState, currentState
+	sourceSnapshotAvailable := sourceState != nil
+	before4, before6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	validateDefaultLocalIPsFromDelta(ctrld.LoggerCtx(ctx, p.logger.Load()), sourceState, transitionID)
+	after4, after6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	p.networkSourceMu.Unlock()
+
+	p.handleDNS64NetworkChange(delta, isMajorChange)
+	validIfaces := networkChangeValidInterfacesFn(ctrld.LoggerCtx(ctx, p.logger.Load()))
+
+	changed := false
+	changedInterface := ""
+	outcome := "ignored"
+	defer func() {
+		p.logger.Load().Debug().Uint64("transition_id", transitionID).
+			Bool("is_major_change", isMajorChange).Bool("changed", changed).
+			Str("interface", changedInterface).
+			Str("default_route_before", delta.Old.DefaultRouteInterface).
+			Str("default_route_after", delta.New.DefaultRouteInterface).
+			Str("source_ipv4_before", before4.String()).Str("source_ipv6_before", before6.String()).
+			Str("source_ipv4_after", after4.String()).
+			Str("source_ipv6_after", after6.String()).
+			Bool("source_snapshot_available", sourceSnapshotAvailable).Str("outcome", outcome).Msg("Network transition")
+	}()
+	activeInterfaceExists := false
+	var changeIPs []netip.Prefix
+	// Check each valid interface for changes
+	for ifaceName := range validIfaces {
+		oldIface, oldExists := delta.Old.Interface[ifaceName]
+		newIface, newExists := delta.New.Interface[ifaceName]
+		if !newExists {
+			continue
+		}
+
+		oldIPs := delta.Old.InterfaceIPs[ifaceName]
+		newIPs := delta.New.InterfaceIPs[ifaceName]
+
+		// if a valid interface did not exist in old
+		// check that its up and has usable IPs
+		if !oldExists {
+			// The interface is new (was not present in the old state).
+			usableNewIPs := filterUsableIPs(newIPs)
+			if newIface.IsUp() && len(usableNewIPs) > 0 {
+				changed = true
+				changeIPs = usableNewIPs
+				changedInterface = ifaceName
+				p.logger.Load().Debug().
+					Str("interface", ifaceName).
+					Interface("new_ips", usableNewIPs).
+					Msg("Interface newly appeared (was not present in old state)")
+				break
+			}
+			continue
+		}
+
+		// Filter new IPs to only those that are usable.
+		usableNewIPs := filterUsableIPs(newIPs)
+
+		// Check if interface is up and has usable IPs.
+		if newIface.IsUp() && len(usableNewIPs) > 0 {
+			activeInterfaceExists = true
+		}
+
+		// Compare interface states and IPs (interfaceIPsEqual will itself filter the IPs).
+		if !interfaceStatesEqual(&oldIface, &newIface) || !interfaceIPsEqual(oldIPs, newIPs) {
+			if newIface.IsUp() && len(usableNewIPs) > 0 {
+				changed = true
+				changeIPs = usableNewIPs
+				changedInterface = ifaceName
+				p.logger.Load().Debug().
+					Str("interface", ifaceName).
+					Interface("old_ips", oldIPs).
+					Interface("new_ips", usableNewIPs).
+					Msg("Interface state or IPs changed")
+				break
+			}
+		}
+	}
+
+	// if the default route changed, set changed to true
+	if delta.New.DefaultRouteInterface != delta.Old.DefaultRouteInterface {
+		changed = true
+		p.logger.Load().Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface)
+	}
+
+	if !changed {
+		p.logger.Load().Debug().Msg("Ignoring interface change - no valid interfaces affected")
+		// Minor interface changes can still accompany pf/WFP or VPN DNS changes.
+		// On macOS, bound the immediate full reconciliation so link-local-only
+		// notification storms do not run pfctl/scutil work for every event.
+		// Windows keeps the existing immediate behavior. Tunnel changes always
+		// bypass the macOS limit, and delayed checks provide a trailing refresh.
+		if dnsIntercept && p.dnsInterceptState != nil {
+			networkChangeIgnoredInterceptFn(p, delta, time.Now())
+		}
+		return
+	}
+
+	if !activeInterfaceExists {
+		outcome = "no_active_interface"
+		p.logger.Load().Debug().Msg("No active interfaces found, skipping reinitialization")
+		return
+	}
+
+	// An ignored event must not cancel this accepted transition. Only a newer
+	// accepted transition can replace its source commit or debounced recovery.
+	p.networkSourceMu.Lock()
+	if p.networkAcceptedGen.Load() > transitionID {
+		p.networkSourceMu.Unlock()
+		outcome = "superseded"
+		return
+	}
+	p.networkAcceptedGen.Store(transitionID)
+	p.networkSourceMu.Unlock()
+
+	p.logger.Load().Debug().Msg("Link state changed, re-bootstrapping")
+	for _, uc := range p.cfg.Upstream {
+		uc.ReBootstrap(ctrld.LoggerCtx(ctx, p.logger.Load()))
+	}
+
+	// Get IPs from default route interface in new state
+	selfIP := networkChangeDefaultRouteIPFn(p)
+
+	// Ensure that selfIP is an IPv4 address.
+	// If defaultRouteIP mistakenly returns an IPv6 (such as a ULA), clear it
+	if ip := net.ParseIP(selfIP); ip != nil && ip.To4() == nil {
+		p.logger.Load().Debug().Msgf("defaultRouteIP returned a non-IPv4 address: %s, ignoring it", selfIP)
+		selfIP = ""
+	}
+	// Route discovery can lag the delta; do not reintroduce an invalid source.
+	if ip := net.ParseIP(selfIP); ip != nil && sourceInvalidReason(delta.New, ip) != "" {
+		selfIP = ""
+	}
+	var ipv6 string
+
+	if delta.New.DefaultRouteInterface != "" {
+		p.logger.Load().Debug().Msgf("default route interface: %s, IPs: %v", delta.New.DefaultRouteInterface, delta.New.InterfaceIPs[delta.New.DefaultRouteInterface])
+		for _, ip := range delta.New.InterfaceIPs[delta.New.DefaultRouteInterface] {
+			if sourceInvalidReason(delta.New, net.ParseIP(ip.Addr().String())) != "" {
+				continue
+			}
+			ipAddr, _ := netip.ParsePrefix(ip.String())
+			addr := ipAddr.Addr()
+			if selfIP == "" && addr.Is4() {
+				p.logger.Load().Debug().Msgf("checking IP: %s", addr.String())
+				if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
+					selfIP = addr.String()
+				}
+			}
+			if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
+				ipv6 = addr.String()
+			}
+		}
+	} else {
+		// If no default route interface is set yet, use the changed IPs
+		p.logger.Load().Debug().Msgf("no default route interface found, using changed IPs: %v", changeIPs)
+		for _, ip := range changeIPs {
+			ipAddr, _ := netip.ParsePrefix(ip.String())
+			addr := ipAddr.Addr()
+			if selfIP == "" && addr.Is4() {
+				p.logger.Load().Debug().Msgf("checking IP: %s", addr.String())
+				if !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
+					selfIP = addr.String()
+				}
+			}
+			if addr.Is6() && !addr.IsLoopback() && !addr.IsLinkLocalUnicast() {
+				ipv6 = addr.String()
+			}
+		}
+	}
+
+	// An ignored event can change source validity without replacing recovery.
+	// Recheck addresses against the newest state after blocking route lookup.
+	p.networkSourceMu.Lock()
+	if p.networkAcceptedGen.Load() != transitionID {
+		p.networkSourceMu.Unlock()
+		outcome = "superseded"
+		return
+	}
+	latestState := p.sourceCommitState(delta)
+	sourceSnapshotAvailable = latestState != nil
+	validateDefaultLocalIPsFromDelta(ctrld.LoggerCtx(ctx, p.logger.Load()), latestState, transitionID)
+	// Only keep candidate addresses that still exist on an up interface.
+	if ip := net.ParseIP(selfIP); ip != nil && ip.To4() != nil && latestState != nil && sourceInvalidReason(latestState, ip) == "" {
+		ctrld.SetDefaultLocalIPv4(ctrld.LoggerCtx(ctx, p.logger.Load()), ip)
+		if !isMobile() && p.ciTable != nil {
+			p.ciTable.SetSelfIP(selfIP)
+		}
+	}
+	if ip := net.ParseIP(ipv6); ip != nil && latestState != nil && sourceInvalidReason(latestState, ip) == "" {
+		ctrld.SetDefaultLocalIPv6(ctrld.LoggerCtx(ctx, p.logger.Load()), ip)
+	}
+	after4, after6 = ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	p.networkSourceMu.Unlock()
+	p.logger.Load().Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6)
+
+	outcome = "accepted"
+	networkChangeReconcileFn(p, ctx, transitionID)
+}
+
+// reconcileNetworkChange is the first DNS/PF mutation boundary after source updates.
+func (p *prog) reconcileNetworkChange(ctx context.Context, transitionID uint64) {
+	p.debounceRecovery(transitionID)
+
+	// Drop stale firewall allowances from the previous network.
+	p.firewallOnNetworkChange()
+
+	// After network changes, verify our pf anchor is still active and
+	// refresh VPN DNS state. Order matters: tunnel checks first (may rebuild
+	// anchor), then VPN DNS refresh (updates exemptions in anchor), then
+	// delayed re-checks for async VPN teardown.
+	if dnsIntercept && p.dnsInterceptState != nil {
+		if !p.pfStabilizing.Load() {
+			p.ensurePFAnchorActive()
+		}
+		// Check tunnel interfaces unconditionally — it decides internally
+		// whether to enter stabilization or rebuild immediately.
+		p.checkTunnelInterfaceChanges()
+		// Refresh VPN DNS routes — runs after tunnel checks so the anchor
+		// rebuild includes current VPN DNS exemptions.
+		if p.vpnDNS != nil {
+			p.vpnDNS.Refresh(ctrld.LoggerCtx(ctx, p.logger.Load()), true)
+		}
+		// Preserve master VM/container forwarded-source reconciliation.
+		p.reconcileForwardedSources()
+		// Schedule delayed re-checks to catch async VPN teardown changes.
+		p.scheduleDelayedRechecks()
+	}
+}
+
 // handleDNSInterceptIgnoredNetworkChange runs the DNS-intercept work for a
 // network delta that did not affect a usable interface. Keeping this path in a
 // method lets tests exercise the callback wiring with synthetic deltas.
@@ -2214,7 +2298,12 @@ const recoveryDebounceWindow = 500 * time.Millisecond
 // recovery runs once with the final network state. All other state updates (IP,
 // pf anchor, VPN DNS, tunnel checks) run immediately — only the recovery flow
 // with its upstream probing and DHCP bypass logic is debounced.
-func (p *prog) debounceRecovery() {
+func (p *prog) debounceRecovery(transitionID uint64) {
+	p.networkSourceMu.Lock()
+	defer p.networkSourceMu.Unlock()
+	if p.networkAcceptedGen.Load() != transitionID {
+		return
+	}
 	p.recoveryDebounceMu.Lock()
 	defer p.recoveryDebounceMu.Unlock()
 
@@ -2223,22 +2312,46 @@ func (p *prog) debounceRecovery() {
 		p.Debug().Msg("Recovery debounce: resetting timer (rapid network change)")
 	}
 	p.recoveryDebounceTimer = time.AfterFunc(recoveryDebounceWindow, func() {
+		p.networkSourceMu.Lock()
+		if p.networkAcceptedGen.Load() != transitionID {
+			p.networkSourceMu.Unlock()
+			return
+		}
 		p.recoveryDebounceMu.Lock()
 		p.recoveryDebounceTimer = nil
 		p.recoveryDebounceMu.Unlock()
-		p.handleRecovery(RecoveryReasonNetworkChange)
+		p.networkSourceMu.Unlock()
+		handleRecoveryForTransitionFn(p, RecoveryReasonNetworkChange, transitionID)
 	})
 	p.Debug().Msg("Recovery debounce: scheduled (500ms window)")
 }
 
 func (p *prog) handleRecovery(reason RecoveryReason) {
-	p.Debug().Msg("Starting recovery process: removing DNS settings")
+	p.handleRecoveryForTransition(reason, 0)
+}
 
-	recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason)
-	if !ok {
-		p.Debug().Msg("Upstream recovery already in progress; skipping duplicate trigger")
+func (p *prog) handleRecoveryForTransition(reason RecoveryReason, transitionID uint64) {
+	p.networkSourceMu.Lock()
+	if transitionID != 0 && p.networkAcceptedGen.Load() != transitionID {
+		p.networkSourceMu.Unlock()
+		p.logger.Load().Debug().Uint64("transition_id", transitionID).Msg("Recovery skipped: transition superseded")
 		return
 	}
+	recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason)
+	p.networkSourceMu.Unlock()
+	if !ok {
+		p.logger.Load().Debug().Uint64("transition_id", transitionID).
+			Uint64("recovery_generation", p.recoveryGen.Load()).
+			Msg("Upstream recovery already in progress; skipping duplicate trigger")
+		return
+	}
+	diagnostic := recoveryDiagnostic{logger: p.logger.Load(), transitionID: transitionID, generation: gen, reason: reason, started: time.Now()}
+	diagnostic.event(p.logger.Load().Debug()).Bool("intercept", interceptRecovery).Msg("Recovery begin")
+	outcome := "superseded"
+	defer func() { diagnostic.end(outcome) }()
+	if reason == RecoveryReasonNetworkChange {
+		p.logger.Load().Debug().Msg("Network change recovery now owns shared recovery state")
+	}
 
 	// For network changes, force-reset all upstream transports synchronously.
 	// The lazy ReBootstrap() called earlier in the network change callback only
@@ -2264,9 +2377,11 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 	}
 
 	upstreams := p.buildRecoveryUpstreams(reason)
-	recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams)
+	recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams, &diagnostic)
 	if err != nil {
-		p.Error().Err(err).Msg("Recovery failed; DNS settings remain removed")
+		if p.recoveryOwnsState(gen) {
+			outcome = "canceled"
+		}
 		p.recoveryCanceledCleanup(gen)
 		return
 	}
@@ -2285,6 +2400,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 		return
 	}
 
+	outcome = "completed"
 	p.Info().Msgf("Recovery completed successfully for upstream %q", recovered)
 }
 
@@ -2337,7 +2453,7 @@ func (p *prog) prepareForRecovery(reason RecoveryReason, interceptRecovery bool)
 	// we do not want to restore any static DNS settings
 	// we must try to get the DHCP values, any static DNS settings
 	// will be appended to nameservers from the saved interface values
-	p.resetDNS(false, false)
+	recoveryResetDNSFn(p, false, false)
 
 	// For an OS failure, reinitialize OS resolver nameservers immediately.
 	if reason == RecoveryReasonOSFailure {
@@ -2398,7 +2514,7 @@ func (p *prog) completeRecoveryWork(reason RecoveryReason, recovered string, int
 
 // waitForUpstreamRecovery checks the provided upstreams concurrently until one recovers.
 // It returns the name of the recovered upstream or an error if the check times out.
-func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string]*ctrld.UpstreamConfig) (string, error) {
+func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string]*ctrld.UpstreamConfig, diagnostic *recoveryDiagnostic) (string, error) {
 	recoveryCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -2434,6 +2550,7 @@ func (p *prog) waitForUpstreamRecovery(ctx context.Context, upstreams map[string
 						}
 						return
 					}
+					diagnostic.failure(err)
 					// Back off the retry cadence for an unroutable endpoint so a
 					// host with IPv6 up but no route to the IPv6 DoH endpoint does
 					// not re-bootstrap/re-check every checkUpstreamBackoffSleep and
@@ -2521,27 +2638,8 @@ func (p *prog) buildRecoveryUpstreams(reason RecoveryReason) map[string]*ctrld.U
 // are still present in the new network state (provided by delta.New).
 // If a stored default IP is no longer active, it resets that default (sets it to nil)
 // so that it won't be used in subsequent custom dialer contexts.
+// Deprecated: retained for callers of the existing exported cli API.
+// The network monitor uses transition-aware validation instead.
 func ValidateDefaultLocalIPsFromDelta(newState *netmon.State) {
-	currentIPv4 := ctrld.GetDefaultLocalIPv4()
-	currentIPv6 := ctrld.GetDefaultLocalIPv6()
-
-	// Build a map of active IP addresses from the new state.
-	activeIPs := make(map[string]bool)
-	for _, prefixes := range newState.InterfaceIPs {
-		for _, prefix := range prefixes {
-			activeIPs[prefix.Addr().String()] = true
-		}
-	}
-
-	// Check if the default IPv4 is still active.
-	if currentIPv4 != nil && !activeIPs[currentIPv4.String()] {
-		mainLog.Load().Debug().Msgf("DefaultLocalIPv4 %s is no longer active in the new state. Resetting.", currentIPv4)
-		ctrld.SetDefaultLocalIPv4(ctrld.LoggerCtx(context.Background(), mainLog.Load()), nil)
-	}
-
-	// Check if the default IPv6 is still active.
-	if currentIPv6 != nil && !activeIPs[currentIPv6.String()] {
-		mainLog.Load().Debug().Msgf("DefaultLocalIPv6 %s is no longer active in the new state. Resetting.", currentIPv6)
-		ctrld.SetDefaultLocalIPv6(ctrld.LoggerCtx(context.Background(), mainLog.Load()), nil)
-	}
+	validateDefaultLocalIPsFromDelta(ctrld.LoggerCtx(context.Background(), mainLog.Load()), newState, 0)
 }

cmd/cli/main.go

@@ -1,13 +1,11 @@
 package cli
 
 import (
-	"encoding/hex"
 	"io"
 	"net"
 	"os"
 	"path/filepath"
 	"sync/atomic"
-	"time"
 
 	"github.com/kardianos/service"
 	"go.uber.org/zap"
@@ -250,17 +248,7 @@ func initCache() {
 // pfProbeSend is a minimal subprocess that sends a pre-built DNS query packet
 // to the specified host on port 53.
 func pfProbeSend(host, hexPacket string) {
-	packet, err := hex.DecodeString(hexPacket)
-	if err != nil {
+	if err := sendPFProbe(host, hexPacket, net.DialTimeout, os.Stdout); err != nil {
 		os.Exit(1)
 	}
-	conn, err := net.DialTimeout("udp", net.JoinHostPort(host, "53"), time.Second)
-	if err != nil {
-		os.Exit(1)
-	}
-	defer conn.Close()
-	conn.SetDeadline(time.Now().Add(time.Second))
-	_, _ = conn.Write(packet)
-	buf := make([]byte, 512)
-	_, _ = conn.Read(buf)
 }

cmd/cli/network_monitor_test.go

@@ -0,0 +1,54 @@
+package cli
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+type sourceTestMonitor struct {
+	callback netmon.ChangeFunc
+	started  bool
+}
+
+func (m *sourceTestMonitor) RegisterChangeCallback(fn netmon.ChangeFunc) func() {
+	m.callback = fn
+	return func() {}
+}
+func (*sourceTestMonitor) IsMajorChangeFrom(*netmon.State, *netmon.State) bool { return true }
+func (m *sourceTestMonitor) Start()                                            { m.started = true }
+
+func TestNetworkMonitorRegistersSourceCleanup(t *testing.T) {
+	sourceTestGlobals(t)
+	old := newNetworkChangeMonitorFn
+	t.Cleanup(func() { newNetworkChangeMonitorFn = old })
+	mon := &sourceTestMonitor{}
+	newNetworkChangeMonitorFn = func(func(string, ...any)) (networkChangeMonitor, error) { return mon, nil }
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(context.Background(), net.ParseIP("2001:db8::10"))
+	calls := 0
+	networkChangeReconcileFn = func(*prog, context.Context, uint64) {
+		calls++
+		if ctrld.GetDefaultLocalIPv4() != nil {
+			t.Fatal("departed IPv4 survived to reconciliation")
+		}
+	}
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})
+	if err := p.monitorNetworkChanges(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if !mon.started || mon.callback == nil {
+		t.Fatal("monitor did not register/start")
+	}
+	mon.callback(&netmon.ChangeDelta{
+		Old: sourceTestState("en0", true, "192.0.2.10/24", "2001:db8::10/64"),
+		New: sourceTestState("en0", true, "2001:db8::10/64"),
+	})
+	if calls != 1 {
+		t.Fatalf("registered callback never reached source reconciliation: %d", calls)
+	}
+}

cmd/cli/network_source_error_test.go

@@ -0,0 +1,34 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestSourceReadFailureDefersAcceptedWrites(t *testing.T) {
+	sourceTestGlobals(t)
+	dnsIntercept = false
+	oldCurrent, oldRead := networkChangeCurrentStateFn, readNetworkSourceStateFn
+	t.Cleanup(func() { networkChangeCurrentStateFn, readNetworkSourceStateFn = oldCurrent, oldRead })
+	major := sourceTestState("en0", true, "192.0.2.10/24")
+	networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return major }
+	readNetworkSourceStateFn = func() (*netmon.State, error) { return nil, errors.New("read failed") }
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP("192.0.2.1"))
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}, networkSourceEpoch: major})
+	calls := 0
+	networkChangeReconcileFn = func(*prog, context.Context, uint64) { calls++ }
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{
+		Monitor: netmon.NewStatic(),
+		Old:     sourceTestState("en0", true, "192.0.2.1/24"), New: major,
+	}, true)
+	if calls != 1 || !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.1")) {
+		t.Fatal("unknown source validity changed a binding or skipped accepted recovery")
+	}
+}

cmd/cli/network_transition.go

@@ -0,0 +1,208 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+	ctrldnet "github.com/Control-D-Inc/ctrld/internal/net"
+)
+
+type networkChangeMonitor interface {
+	RegisterChangeCallback(netmon.ChangeFunc) func()
+	IsMajorChangeFrom(*netmon.State, *netmon.State) bool
+	Start()
+}
+
+// Tests replace these functions to avoid changes to host DNS and PF state.
+var (
+	newNetworkChangeMonitorFn = func(logf func(string, ...any)) (networkChangeMonitor, error) {
+		return netmon.New(logf)
+	}
+	networkChangeCurrentStateFn = func(delta *netmon.ChangeDelta) *netmon.State {
+		if delta.Monitor != nil {
+			return delta.Monitor.InterfaceState()
+		}
+		return delta.New
+	}
+	readNetworkSourceStateFn        = readNetworkSourceState
+	networkChangeValidInterfacesFn  = ctrld.ValidInterfaces
+	networkChangeDefaultRouteIPFn   = (*prog).defaultRouteIP
+	networkChangeReconcileFn        = (*prog).reconcileNetworkChange
+	networkChangeIgnoredInterceptFn = (*prog).handleDNSInterceptIgnoredNetworkChange
+	handleRecoveryForTransitionFn   = (*prog).handleRecoveryForTransition
+	recoveryResetDNSFn              = (*prog).resetDNS
+)
+
+// Minor callbacks have no observation sequence in netmon v1.74.0. Read only
+// current interface addresses and flags rather than trusting their arrival order.
+func readNetworkSourceState() (*netmon.State, error) {
+	interfaces, err := net.Interfaces()
+	if err != nil {
+		return nil, err
+	}
+	state := &netmon.State{Interface: map[string]netmon.Interface{}, InterfaceIPs: map[string][]netip.Prefix{}}
+	for _, iface := range interfaces {
+		addresses, err := iface.Addrs()
+		if err != nil {
+			return nil, err
+		}
+		state.Interface[iface.Name] = netmon.Interface{Interface: &iface}
+		for _, address := range addresses {
+			var ip net.IP
+			switch address := address.(type) {
+			case *net.IPNet:
+				ip = address.IP
+			case *net.IPAddr:
+				ip = address.IP
+			default:
+				return nil, errors.New("unsupported interface address type")
+			}
+			addr, ok := netip.AddrFromSlice(ip)
+			if !ok {
+				return nil, errors.New("invalid interface address")
+			}
+			addr = addr.Unmap()
+			state.InterfaceIPs[iface.Name] = append(state.InterfaceIPs[iface.Name], netip.PrefixFrom(addr, addr.BitLen()))
+		}
+	}
+	return state, nil
+}
+
+// netmon v1.74.0 caches only major snapshots. Minor and time-jump events
+// belong to their Old snapshot, not their distinct New snapshot.
+func networkSnapshotCurrent(delta *netmon.ChangeDelta, major bool, current *netmon.State) bool {
+	if major || delta.Monitor == nil {
+		return current == delta.New
+	}
+	return current == delta.Old
+}
+
+// Callers hold networkSourceMu. Preserve minor changes within the current
+// major epoch, but use a new major snapshot even before its callback runs.
+func (p *prog) sourceCommitState(delta *netmon.ChangeDelta) *netmon.State {
+	if delta.Monitor != nil {
+		if current := networkChangeCurrentStateFn(delta); current != p.networkSourceEpoch {
+			return current
+		}
+	}
+	return p.networkSourceState
+}
+
+// sourceInvalidReason considers all up interfaces, not just hardware ports: a
+// preferred source remains valid if it moved to another active interface.
+func sourceInvalidReason(state *netmon.State, ip net.IP) string {
+	if ip == nil || state == nil {
+		return ""
+	}
+	reason := "address_removed"
+	for name, prefixes := range state.InterfaceIPs {
+		for _, prefix := range prefixes {
+			if !ip.Equal(net.ParseIP(prefix.Addr().String())) {
+				continue
+			}
+			iface, exists := state.Interface[name]
+			if exists && iface.IsUp() {
+				return ""
+			}
+			if exists {
+				reason = "interface_down"
+			} else if reason != "interface_down" {
+				reason = "interface_removed"
+			}
+		}
+	}
+	return reason
+}
+
+func validateDefaultLocalIPsFromDelta(ctx context.Context, state *netmon.State, transitionID uint64) {
+	before4, before6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	reason4, reason6 := sourceInvalidReason(state, before4), sourceInvalidReason(state, before6)
+	if reason4 != "" {
+		ctrld.SetDefaultLocalIPv4(ctx, nil)
+	}
+	if reason6 != "" {
+		ctrld.SetDefaultLocalIPv6(ctx, nil)
+	}
+	if reason4 == "" && reason6 == "" {
+		return
+	}
+	// One warning per actual invalidation, not per notification or query. The
+	// normal warning buffer retains this summary when debug traffic rotates.
+	ctrld.LoggerFromCtx(ctx).Warn().Uint64("transition_id", transitionID).
+		Str("default_route", state.DefaultRouteInterface).
+		Str("source_ipv4_before", before4.String()).Str("source_ipv6_before", before6.String()).
+		Str("source_ipv4_after", ctrld.GetDefaultLocalIPv4().String()).
+		Str("source_ipv6_after", ctrld.GetDefaultLocalIPv6().String()).
+		Str("ipv4_clear_reason", reason4).Str("ipv6_clear_reason", reason6).
+		Msg("Removed stale resolver source")
+}
+
+type recoveryDiagnostic struct {
+	logger       *ctrld.Logger
+	transitionID uint64
+	generation   uint64
+	reason       RecoveryReason
+	started      time.Time
+	firstFailure sync.Once
+	failed       atomic.Bool
+}
+
+func recoveryReasonName(reason RecoveryReason) string {
+	switch reason {
+	case RecoveryReasonNetworkChange:
+		return "network_change"
+	case RecoveryReasonRegularFailure:
+		return "upstream_failure"
+	case RecoveryReasonOSFailure:
+		return "os_resolver_failure"
+	default:
+		return "unknown"
+	}
+}
+
+func (d *recoveryDiagnostic) event(e *ctrld.LogEvent) *ctrld.LogEvent {
+	return e.Uint64("transition_id", d.transitionID).Uint64("recovery_generation", d.generation).
+		Str("recovery_reason", recoveryReasonName(d.reason))
+}
+
+func (d *recoveryDiagnostic) failure(err error) {
+	d.firstFailure.Do(func() {
+		d.failed.Store(true)
+		// Classify only; resolver errors can contain queried names or endpoints.
+		failure := "upstream_error"
+		var networkError net.Error
+		if errors.Is(err, syscall.EADDRNOTAVAIL) {
+			failure = "source_unavailable"
+		} else if ctrldnet.IsUnreachable(err) {
+			failure = "network_unreachable"
+		} else if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &networkError) && networkError.Timeout()) {
+			failure = "timeout"
+		}
+		d.event(d.log().Warn()).Str("failure", failure).
+			Msg("Recovery waiting for upstream; first failure")
+	})
+}
+
+func (d *recoveryDiagnostic) end(outcome string) {
+	e := d.log().Debug()
+	if outcome == "canceled" || d.failed.Load() {
+		e = d.log().Warn()
+	}
+	d.event(e).Str("outcome", outcome).Bool("had_failure", d.failed.Load()).Int64("duration_ms", time.Since(d.started).Milliseconds()).Msg("Recovery end")
+}
+
+func (d *recoveryDiagnostic) log() *ctrld.Logger {
+	if d.logger != nil {
+		return d.logger
+	}
+	return mainLog.Load()
+}

cmd/cli/network_transition_concurrency_test.go

@@ -0,0 +1,89 @@
+package cli
+
+import (
+	"context"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"tailscale.com/net/netmon"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+)
+
+func TestNetworkChangeOverlappingCallbackCannotRestoreRemovedSource(t *testing.T) {
+	sourceTestGlobals(t)
+	logs := captureTransitionLogs(t)
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	networkChangeDefaultRouteIPFn = func(*prog) string {
+		close(entered)
+		<-release
+		return "192.0.2.10"
+	}
+	reconciliations := 0
+	networkChangeReconcileFn = func(*prog, context.Context, uint64) { reconciliations++ }
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) {}
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})
+	stateA := sourceTestState("en0", true, "192.0.2.10/24")
+	t.Cleanup(func() {
+		select {
+		case <-release:
+		default:
+			close(release)
+		}
+		<-done
+	})
+	go func() {
+		defer close(done)
+		p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.20/24"), New: stateA}, true)
+	}()
+	select {
+	case <-entered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("first callback did not reach route discovery")
+	}
+	// The newer callback completes while A is paused. Its ignored/address-loss
+	// path clears the source and deliberately schedules no replacement recovery.
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: stateA, New: sourceTestState("en0", true)}, true)
+	close(release)
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("superseded callback did not finish")
+	}
+	if ctrld.GetDefaultLocalIPv4() != nil {
+		t.Fatal("older callback restored the removed source")
+	}
+	if reconciliations != 1 || !strings.Contains(logs.String(), `"outcome":"accepted"`) {
+		t.Fatal("an ignored address-loss delta must preserve the accepted recovery")
+	}
+}
+
+func TestNetworkChangeLateOldSnapshotIsRejected(t *testing.T) {
+	sourceTestGlobals(t)
+	old := networkChangeCurrentStateFn
+	t.Cleanup(func() { networkChangeCurrentStateFn = old })
+	latest := sourceTestState("en0", true)
+	networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return latest }
+	ctrld.SetDefaultLocalIPv4(context.Background(), nil)
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	networkChangeReconcileFn = func(*prog, context.Context, uint64) { t.Fatal("old snapshot reconciled") }
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { t.Fatal("old snapshot reconciled") }
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: latest, New: sourceTestState("en0", true, "192.0.2.10/24")}, true)
+	if p.networkTransitionGen.Load() != 0 || ctrld.GetDefaultLocalIPv4() != nil {
+		t.Fatal("old snapshot became a new transition/source")
+	}
+}
+
+func TestRecoveryRejectsSupersededTransition(t *testing.T) {
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	p.networkAcceptedGen.Store(2)
+	p.handleRecoveryForTransition(RecoveryReasonNetworkChange, 1)
+	if p.recoveryGen.Load() != 0 || p.recoveryRunning.Load() {
+		t.Fatal("superseded transition started recovery")
+	}
+}

cmd/cli/network_transition_reordered_test.go

@@ -0,0 +1,96 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestLateSameEpochCallbacksCannotRestoreRemovedSource(t *testing.T) {
+	for _, initial := range []string{"192.0.2.1", "192.0.2.10"} {
+		t.Run(initial, func(t *testing.T) {
+			sourceTestGlobals(t)
+			dnsIntercept = false
+			oldCurrent, oldRead := networkChangeCurrentStateFn, readNetworkSourceStateFn
+			t.Cleanup(func() { networkChangeCurrentStateFn, readNetworkSourceStateFn = oldCurrent, oldRead })
+			mon := netmon.NewStatic()
+			before := sourceTestState("en0", true, "192.0.2.1/24")
+			major := sourceTestState("ipsec0", true, "192.0.2.1/24")
+			major.Interface["ipsec0"] = netmon.Interface{Interface: &net.Interface{Name: "ipsec0", Flags: net.FlagUp | net.FlagPointToPoint}}
+			major.InterfaceIPs["ipsec0"] = []netip.Prefix{netip.MustParsePrefix("192.0.2.10/24")}
+			latest := sourceTestState("ipsec0", true, "192.0.2.1/24")
+			latest.Interface["ipsec0"] = major.Interface["ipsec0"]
+			// netmon ignores ipsec address changes on Darwin. Its major cache
+			// stays at A while B removes the source. These are OS-read seams.
+			networkChangeCurrentStateFn = func(*netmon.ChangeDelta) *netmon.State { return major }
+			reads := 0
+			readNetworkSourceStateFn = func() (*netmon.State, error) { reads++; return latest, nil }
+			ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP(initial))
+			ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+			p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+			reconciled := 0
+			networkChangeReconcileFn = func(*prog, context.Context, uint64) { reconciled++ }
+			// B finishes before A's major callback even enters. A second, older
+			// minor callback also arrives late in that same cache epoch.
+			p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Monitor: mon, Old: major, New: latest}, false)
+			olderMinor := *major
+			p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Monitor: mon, Old: major, New: &olderMinor}, false)
+			if len(p.networkSourceState.InterfaceIPs["ipsec0"]) != 0 {
+				t.Fatal("late minor callback replaced current source validity")
+			}
+			p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Monitor: mon, Old: before, New: major}, true)
+			want := net.IP(nil)
+			if initial == "192.0.2.1" {
+				want = net.ParseIP(initial)
+			}
+			if !ctrld.GetDefaultLocalIPv4().Equal(want) {
+				t.Fatalf("late major restored an obsolete source: got %v want %v", ctrld.GetDefaultLocalIPv4(), want)
+			}
+			if reads != 3 || reconciled != 1 {
+				t.Fatalf("fresh reads=%d reconciliation=%d", reads, reconciled)
+			}
+		})
+	}
+}
+
+func TestMinorSourceReadFailureDoesNotSkipReconciliation(t *testing.T) {
+	sourceTestGlobals(t)
+	oldRead := readNetworkSourceStateFn
+	t.Cleanup(func() { readNetworkSourceStateFn = oldRead })
+	readNetworkSourceStateFn = func() (*netmon.State, error) { return nil, errors.New("read failed") }
+	mon := netmon.NewStatic()
+	old := mon.InterfaceState()
+	next := *old
+	networkChangeValidInterfacesFn = func(context.Context) map[string]struct{} { return map[string]struct{}{} }
+	calls := 0
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ }
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Monitor: mon, Old: old, New: &next}, false)
+	if calls != 1 || p.networkSourceState != nil || !p.networkSourceReadFailed {
+		t.Fatal("source-read failure skipped reconciliation or treated unknown data as current")
+	}
+}
+
+func TestReadNetworkSourceState(t *testing.T) {
+	state, err := readNetworkSourceState()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for name, iface := range state.Interface {
+		if iface.Interface == nil || iface.Name != name {
+			t.Fatal("interface snapshots share the wrong object")
+		}
+		for _, prefix := range state.InterfaceIPs[name] {
+			if !prefix.IsValid() {
+				t.Fatalf("invalid source prefix on %s", name)
+			}
+		}
+	}
+}

cmd/cli/network_transition_review_test.go

@@ -0,0 +1,130 @@
+package cli
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func TestNetmonMinorAndWakeSnapshotReconciliation(t *testing.T) {
+	for _, timeJump := range []bool{false, true} {
+		t.Run(map[bool]string{false: "minor", true: "time_jump"}[timeJump], func(t *testing.T) {
+			sourceTestGlobals(t)
+			mon := netmon.NewStatic()
+			// NewStatic has no monitoring resources to close.
+			old := mon.InterfaceState()
+			next := *old
+			// An identical snapshot is non-major; NewStatic has no OS monitor for IsMajorChangeFrom.
+			major := false
+			networkChangeValidInterfacesFn = func(context.Context) map[string]struct{} { return map[string]struct{}{} }
+			calls := 0
+			networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ }
+			p := sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})
+			delta := &netmon.ChangeDelta{Monitor: mon, Old: old, New: &next, TimeJumped: timeJump}
+			//lint:ignore SA1019 This fixture reproduces netmon v1.74.0's legacy wake flag.
+			delta.Major = timeJump
+			p.handleNetworkChange(context.Background(), delta, major)
+			if calls != 1 {
+				t.Fatalf("real netmon non-major delta discarded: ignored-path calls=%d transition ID=%d", calls, p.networkTransitionGen.Load())
+			}
+		})
+	}
+}
+
+func TestIgnoredDeltaPreservesAcceptedRecovery(t *testing.T) {
+	sourceTestGlobals(t)
+	dnsIntercept = false
+	oldRecovery := handleRecoveryForTransitionFn
+	t.Cleanup(func() { handleRecoveryForTransitionFn = oldRecovery })
+	calls := make(chan uint64, 2)
+	handleRecoveryForTransitionFn = func(_ *prog, _ RecoveryReason, id uint64) { calls <- id }
+	networkChangeReconcileFn = (*prog).reconcileNetworkChange // Real accepted-path wiring, with recovery stubbed below.
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true)
+	b := sourceTestState("en0", true, "192.0.2.20/24")
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: a, New: b}, false)
+	select {
+	case <-calls:
+	case <-time.After(1200 * time.Millisecond):
+		t.Fatal("ignored delta canceled accepted recovery; no replacement was scheduled")
+	}
+}
+
+func TestIgnoredDeltaPreservesValidPendingSource(t *testing.T) {
+	sourceTestGlobals(t)
+	dnsIntercept = false
+	ctrld.SetDefaultLocalIPv4(context.Background(), nil)
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	networkChangeDefaultRouteIPFn = func(*prog) string { close(entered); <-release; return "192.0.2.20" }
+	reconciled := make(chan uint64, 2)
+	networkChangeReconcileFn = func(_ *prog, _ context.Context, id uint64) { reconciled <- id }
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	go func() {
+		defer close(done)
+		p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true)
+	}()
+	<-entered
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: a, New: sourceTestState("en0", true, "192.0.2.20/24")}, false)
+	close(release)
+	<-done
+	if !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.20")) || len(reconciled) != 1 {
+		t.Fatalf("valid source discarded after ignored delta: source=%v reconciliations=%d", ctrld.GetDefaultLocalIPv4(), len(reconciled))
+	}
+}
+
+func TestNewAcceptedDeltaSupersedesPendingAcceptedSource(t *testing.T) {
+	sourceTestGlobals(t)
+	dnsIntercept = false
+	entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	var lookups atomic.Int32
+	networkChangeDefaultRouteIPFn = func(*prog) string {
+		if lookups.Add(1) == 1 {
+			close(entered)
+			<-release
+			return "192.0.2.20"
+		}
+		return ""
+	}
+	reconciled := make(chan uint64, 2)
+	networkChangeReconcileFn = func(_ *prog, _ context.Context, id uint64) { reconciled <- id }
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	go func() {
+		defer close(done)
+		p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true)
+	}()
+	<-entered
+	p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: a, New: sourceTestState("en0", true, "192.0.2.30/24")}, true)
+	close(release)
+	<-done
+	if !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.30")) || len(reconciled) != 1 || <-reconciled != 2 {
+		t.Fatal("older accepted delta replaced the newer source or recovery")
+	}
+}
+
+func TestSourceCommitUsesCurrentMajorEpoch(t *testing.T) {
+	mon := netmon.NewStatic()
+	latest := mon.InterfaceState()
+	minor := *latest
+	p := sourceTestProg(&prog{networkSourceState: &minor, networkSourceEpoch: latest})
+	delta := &netmon.ChangeDelta{Monitor: mon, Old: latest, New: &minor}
+	if p.sourceCommitState(delta) != &minor {
+		t.Fatal("discarded the observed minor update within the current epoch")
+	}
+	p.networkSourceEpoch = &minor
+	if p.sourceCommitState(delta) != latest {
+		t.Fatal("did not refresh from a newer major epoch")
+	}
+	if networkSnapshotCurrent(&netmon.ChangeDelta{Monitor: mon, Old: &minor, New: &minor}, false, latest) {
+		t.Fatal("accepted a minor callback from an obsolete major epoch")
+	}
+}

cmd/cli/network_transition_test.go

@@ -0,0 +1,327 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"net/netip"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"tailscale.com/net/netmon"
+
+	"github.com/Control-D-Inc/ctrld"
+)
+
+func sourceTestState(route string, up bool, ips ...string) *netmon.State {
+	flags := net.Flags(0)
+	if up {
+		flags = net.FlagUp
+	}
+	s := &netmon.State{
+		DefaultRouteInterface: route,
+		Interface:             map[string]netmon.Interface{"en0": {Interface: &net.Interface{Name: "en0", Flags: flags}}},
+		InterfaceIPs:          map[string][]netip.Prefix{},
+	}
+	for _, ip := range ips {
+		s.InterfaceIPs["en0"] = append(s.InterfaceIPs["en0"], netip.MustParsePrefix(ip))
+	}
+	return s
+}
+
+func captureTransitionLogs(t *testing.T) *syncBuffer {
+	t.Helper()
+	old := mainLog.Load()
+	buf := &syncBuffer{}
+	logger := sourceTestLogger(buf)
+	mainLog.Store(logger)
+	t.Cleanup(func() { mainLog.Store(old) })
+	return buf
+}
+
+func sourceTestGlobals(t *testing.T) {
+	t.Helper()
+	v4, v6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	valid, route := networkChangeValidInterfacesFn, networkChangeDefaultRouteIPFn
+	reconcile, ignored := networkChangeReconcileFn, networkChangeIgnoredInterceptFn
+	intercept := dnsIntercept
+	dnsIntercept = true
+	networkChangeValidInterfacesFn = func(context.Context) map[string]struct{} { return map[string]struct{}{"en0": {}} }
+	networkChangeDefaultRouteIPFn = func(*prog) string { return "" }
+	t.Cleanup(func() {
+		ctrld.SetDefaultLocalIPv4(context.Background(), v4)
+		ctrld.SetDefaultLocalIPv6(context.Background(), v6)
+		networkChangeValidInterfacesFn, networkChangeDefaultRouteIPFn = valid, route
+		networkChangeReconcileFn, networkChangeIgnoredInterceptFn = reconcile, ignored
+		dnsIntercept = intercept
+	})
+}
+
+func TestNetworkChangeSourceLifecycle(t *testing.T) {
+	const v4, v6 = "192.0.2.10", "2001:db8::10"
+	for _, tt := range []struct {
+		name                      string
+		newState                  *netmon.State
+		preferred                 string
+		want4, want6              string
+		outcome, reason4, reason6 string
+	}{
+		{"dhcp_lease_replacement_keeps_ipv6", sourceTestState("en0", true, "192.0.2.20/24", v6+"/64"), "", "192.0.2.20", v6, "accepted", "address_removed", ""},
+		{"dhcp_lease_replacement", sourceTestState("en0", true, "192.0.2.20/24"), "", "192.0.2.20", "", "accepted", "address_removed", "address_removed"},
+		{"accepted_ipv4_loss", sourceTestState("en0", true, v6+"/64"), "", "", v6, "accepted", "address_removed", ""},
+		{"accepted_ipv6_loss", sourceTestState("en0", true, v4+"/24"), v4, v4, "", "accepted", "", "address_removed"},
+		{"stale_route_discovery", sourceTestState("en0", true, v6+"/64"), v4, "", v6, "accepted", "address_removed", ""},
+		{"ignored_down_retains_addresses", sourceTestState("en0", false, v4+"/24", v6+"/64"), "", "", "", "ignored", "interface_down", "interface_down"},
+		{"ignored_address_loss", sourceTestState("en0", true), "", "", "", "ignored", "address_removed", "address_removed"},
+		{"skipped_no_active_interface", sourceTestState("", false, v4+"/24", v6+"/64"), "", "", "", "no_active_interface", "interface_down", "interface_down"},
+		{"ignored_valid_sources", sourceTestState("en0", true, v4+"/24", v6+"/64"), "", v4, v6, "ignored", "", ""},
+		{"accepted_valid_preferred_source", sourceTestState("en0", true, "192.0.2.20/24", v4+"/24", v6+"/64"), v4, v4, v6, "accepted", "", ""},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			sourceTestGlobals(t)
+			logs := captureTransitionLogs(t)
+			ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP(v4))
+			ctrld.SetDefaultLocalIPv6(context.Background(), net.ParseIP(v6))
+			networkChangeDefaultRouteIPFn = func(*prog) string { return tt.preferred }
+			p := sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})
+			accepted, ignored := 0, 0
+			assertSources := func() {
+				t.Helper()
+				if got := ctrld.GetDefaultLocalIPv4(); !got.Equal(net.ParseIP(tt.want4)) {
+					t.Errorf("IPv4 = %v, want %s", got, tt.want4)
+				}
+				if got := ctrld.GetDefaultLocalIPv6(); !got.Equal(net.ParseIP(tt.want6)) {
+					t.Errorf("IPv6 = %v, want %s", got, tt.want6)
+				}
+			}
+			// These seams precede all DNS/PF writes and recovery timers, on every OS.
+			networkChangeReconcileFn = func(_ *prog, _ context.Context, id uint64) {
+				accepted++
+				if id != 1 {
+					t.Errorf("transition ID = %d", id)
+				}
+				assertSources()
+			}
+			networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { ignored++; assertSources() }
+			p.handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: sourceTestState("en0", true, v4+"/24", v6+"/64"), New: tt.newState}, true)
+			assertSources()
+			wantAccepted, wantIgnored := 0, 0
+			if tt.outcome == "accepted" {
+				wantAccepted = 1
+			}
+			if tt.outcome == "ignored" {
+				wantIgnored = 1
+			}
+			if accepted != wantAccepted || ignored != wantIgnored {
+				t.Fatalf("mutation boundaries: accepted=%d ignored=%d; want %d/%d", accepted, ignored, wantAccepted, wantIgnored)
+			}
+			var transition, warning map[string]any
+			for _, line := range strings.Split(strings.TrimSpace(logs.String()), "\n") {
+				var event map[string]any
+				if err := json.Unmarshal([]byte(line), &event); err != nil {
+					t.Fatal(err)
+				}
+				switch event["message"] {
+				case "Network transition":
+					transition = event
+				case "Removed stale resolver source":
+					warning = event
+				}
+			}
+			if transition["outcome"] != tt.outcome || transition["transition_id"] != float64(1) {
+				t.Fatalf("transition: %v", transition)
+			}
+			wantWarning := tt.reason4 != "" || tt.reason6 != ""
+			if (warning != nil) != wantWarning {
+				t.Fatalf("warning = %v, want warning %v", warning, wantWarning)
+			}
+			if wantWarning && (warning["ipv4_clear_reason"] != tt.reason4 || warning["ipv6_clear_reason"] != tt.reason6 || warning["level"] != "warn") {
+				t.Fatalf("warning reasons/level: %v", warning)
+			}
+		})
+	}
+}
+
+func TestNetworkChangeDoesNotRestoreDownDefaultRouteSources(t *testing.T) {
+	sourceTestGlobals(t)
+	const v4, v6 = "192.0.2.10", "2001:db8::10"
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP(v4))
+	ctrld.SetDefaultLocalIPv6(context.Background(), net.ParseIP(v6))
+	networkChangeDefaultRouteIPFn = func(*prog) string { return v4 }
+	old := sourceTestState("en1", true, "2001:db8:1::10/64")
+	old.Interface["en1"] = netmon.Interface{Interface: &net.Interface{Name: "en1", Flags: net.FlagUp}}
+	old.InterfaceIPs["en1"] = []netip.Prefix{netip.MustParsePrefix(v4 + "/24"), netip.MustParsePrefix(v6 + "/64")}
+	newState := sourceTestState("en1", true, "2001:db8:1::20/64")
+	newState.Interface["en1"] = netmon.Interface{Interface: &net.Interface{Name: "en1"}}
+	newState.InterfaceIPs["en1"] = old.InterfaceIPs["en1"]
+	called := 0
+	networkChangeReconcileFn = func(_ *prog, _ context.Context, _ uint64) {
+		called++
+		if ctrld.GetDefaultLocalIPv4() != nil || ctrld.GetDefaultLocalIPv6() != nil {
+			t.Error("down-interface sources reintroduced before recovery")
+		}
+	}
+	networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { t.Fatal("accepted delta was ignored") }
+	(sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})).handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: old, New: newState}, true)
+	if called != 1 {
+		t.Fatalf("reconciliation called %d times", called)
+	}
+}
+
+func TestNetworkChangeSourceMovedToUpInterface(t *testing.T) {
+	sourceTestGlobals(t)
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	state := sourceTestState("en0", false, "192.0.2.10/24")
+	state.Interface["utun0"] = netmon.Interface{Interface: &net.Interface{Name: "utun0", Flags: net.FlagUp}}
+	state.InterfaceIPs["utun0"] = state.InterfaceIPs["en0"]
+	called := 0
+	networkChangeReconcileFn = func(_ *prog, _ context.Context, _ uint64) { t.Fatal("unexpected recovery") }
+	networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { called++ }
+	(sourceTestProg(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}})).handleNetworkChange(context.Background(), &netmon.ChangeDelta{Old: state, New: state}, false)
+	if called != 1 || !ctrld.GetDefaultLocalIPv4().Equal(net.ParseIP("192.0.2.10")) {
+		t.Fatal("valid source on another up interface was cleared")
+	}
+}
+
+func TestRecoveryTransitionDiagnostics(t *testing.T) {
+	logs := captureTransitionLogs(t)
+	original, intercept := recoveryResetDNSFn, dnsIntercept
+	dnsIntercept = false
+	t.Cleanup(func() { recoveryResetDNSFn = original; dnsIntercept = intercept })
+	for _, superseded := range []bool{false, true} {
+		p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+		calls := 0
+		recoveryResetDNSFn = func(p *prog, start, restore bool) {
+			calls++
+			if start || restore {
+				t.Error("changed reset flags")
+			}
+			if superseded {
+				p.beginRecovery(RecoveryReasonNetworkChange)
+			} else {
+				p.recoveryCancel()
+			}
+		}
+		p.networkAcceptedGen.Store(42)
+		p.handleRecoveryForTransition(RecoveryReasonNetworkChange, 42)
+		if calls != 1 {
+			t.Fatalf("resetDNS seam called %d times", calls)
+		}
+		if p.recoveryCancel != nil {
+			p.recoveryCancel()
+		}
+	}
+	for _, expected := range []string{`"transition_id":42`, `"recovery_generation":1`, `"message":"Recovery begin"`, `"outcome":"canceled"`, `"outcome":"superseded"`, `"message":"Recovery end"`} {
+		if !strings.Contains(logs.String(), expected) {
+			t.Errorf("missing %s in %s", expected, logs.String())
+		}
+	}
+}
+
+func TestRecoveryDebouncePreservesTransitionID(t *testing.T) {
+	original := handleRecoveryForTransitionFn
+	t.Cleanup(func() { handleRecoveryForTransitionFn = original })
+	ids := make(chan uint64, 2)
+	handleRecoveryForTransitionFn = func(_ *prog, reason RecoveryReason, id uint64) {
+		if reason != RecoveryReasonNetworkChange {
+			t.Error("wrong recovery reason")
+		}
+		ids <- id
+	}
+	p := sourceTestProg(&prog{cfg: &ctrld.Config{}})
+	p.networkAcceptedGen.Store(11)
+	p.debounceRecovery(11)
+	p.networkAcceptedGen.Store(12)
+	p.debounceRecovery(12)
+	p.debounceRecovery(11) // a delayed older callback must not replace the new timer
+	select {
+	case id := <-ids:
+		if id != 12 {
+			t.Fatalf("recovery transition = %d, want 12", id)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("debounced recovery did not run")
+	}
+}
+
+func TestTransitionWarningsSurviveDebugRotation(t *testing.T) {
+	sourceTestGlobals(t)
+	old := mainLog.Load()
+	debug, warnings := newLogWriterWithSize(2048), newSmallLogWriter()
+	// Runtime summaries occur after initialization, outside the preserved prefix.
+	debug.Write([]byte(logWriterInitEndMarker))
+	config := zap.NewProductionEncoderConfig()
+	config.MessageKey = "message"
+	logger := &ctrld.Logger{Logger: zap.New(zapcore.NewTee(
+		zapcore.NewCore(zapcore.NewJSONEncoder(config), zapcore.AddSync(debug), zap.DebugLevel),
+		zapcore.NewCore(zapcore.NewJSONEncoder(config), zapcore.AddSync(warnings), zap.WarnLevel),
+	))}
+	mainLog.Store(logger)
+	t.Cleanup(func() { mainLog.Store(old) })
+	ctrld.SetDefaultLocalIPv4(context.Background(), net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(context.Background(), nil)
+	state := sourceTestState("en0", false, "192.0.2.10/24")
+	for range 100 {
+		validateDefaultLocalIPsFromDelta(ctrld.LoggerCtx(context.Background(), mainLog.Load()), state, 7)
+	}
+	d := &recoveryDiagnostic{transitionID: 7, generation: 9, started: time.Now()}
+	var wg sync.WaitGroup
+	for range 100 {
+		wg.Add(1)
+		go func() { defer wg.Done(); d.failure(errors.New("private.customer.example secret payload")) }()
+	}
+	wg.Wait()
+	d.end("completed")
+	(&recoveryDiagnostic{transitionID: 8, generation: 10, started: time.Now()}).end("completed")
+	for range 100 {
+		logger.Debug().Msg(strings.Repeat("noise", 100))
+	}
+	text := warnings.buf.String()
+	for _, msg := range []string{"Removed stale resolver source", "Recovery waiting for upstream; first failure"} {
+		if strings.Count(text, msg) != 1 {
+			t.Fatalf("expected one retained %q: %s", msg, text)
+		}
+		if strings.Contains(debug.buf.String(), msg) {
+			t.Fatalf("debug buffer did not rotate %q", msg)
+		}
+	}
+	if strings.Count(text, "Recovery end") != 1 || !strings.Contains(text, `"outcome":"completed"`) {
+		t.Fatal("recovery completion must survive rotation beside its retained failure")
+	}
+	if strings.Contains(text, "private.customer") || strings.Contains(text, "secret payload") {
+		t.Fatalf("raw payload or normal debug elevated: %s", text)
+	}
+}
+
+func TestRecoveryWaitLogsFirstFailure(t *testing.T) {
+	logs := captureTransitionLogs(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	d := &recoveryDiagnostic{transitionID: 8, generation: 2, started: time.Now()}
+	// Invalid resolver type fails construction before any network access.
+	_, err := (sourceTestProg(&prog{})).waitForUpstreamRecovery(ctx, map[string]*ctrld.UpstreamConfig{"invalid": {Type: "invalid"}}, d)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("wait result = %v", err)
+	}
+	if strings.Count(logs.String(), "Recovery waiting for upstream; first failure") != 1 {
+		t.Fatalf("missing first-failure summary: %s", logs.String())
+	}
+}
+
+// sourceTestProg adapts the source fixtures to master's per-program logger.
+func sourceTestProg(p *prog) *prog {
+	p.logger.Store(mainLog.Load())
+	return p
+}
+
+func sourceTestLogger(buf *syncBuffer) *ctrld.Logger {
+	config := zap.NewProductionEncoderConfig()
+	config.MessageKey = "message"
+	return &ctrld.Logger{Logger: zap.New(zapcore.NewCore(zapcore.NewJSONEncoder(config), zapcore.AddSync(buf), zap.DebugLevel))}
+}

cmd/cli/pf_probe.go

@@ -0,0 +1,233 @@
+package cli
+
+import (
+	"bufio"
+	"context"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/netip"
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"tailscale.com/net/tsaddr"
+)
+
+type pfProbeResult uint8
+
+const (
+	pfProbeIndeterminate pfProbeResult = iota
+	pfProbeIntercepted
+	pfProbeNotIntercepted
+)
+
+func (r pfProbeResult) String() string {
+	switch r {
+	case pfProbeIntercepted:
+		return "intercepted"
+	case pfProbeNotIntercepted:
+		return "not_intercepted"
+	default:
+		return "indeterminate"
+	}
+}
+
+type pfProbeObservation struct {
+	result             pfProbeResult
+	stage              string
+	code               string
+	probeID            string
+	recoveryGeneration uint64
+	target             string
+}
+
+// Conditions exclude attempt IDs and generations so repeated failures stay bounded.
+type pfProbeCondition struct {
+	result              pfProbeResult
+	stage, code, target string
+}
+
+type pfProbeLogState struct {
+	mu          sync.Mutex
+	previous    pfProbeCondition
+	initialized bool
+	repeats     uint64
+}
+
+// change returns whether to retain a transition and the preceding repeat count.
+// A restored receipt is retained too, so an old warning does not imply an outage.
+func (s *pfProbeLogState) change(o pfProbeObservation) (retain bool, repeats uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	condition := pfProbeCondition{o.result, o.stage, o.code, o.target}
+	if s.initialized && condition == s.previous {
+		s.repeats++
+		return false, s.repeats
+	}
+	retain = o.result != pfProbeIntercepted || (s.initialized && s.previous.result != pfProbeIntercepted)
+	repeats = s.repeats
+	s.previous, s.initialized, s.repeats = condition, true, 0
+	return retain, repeats
+}
+
+// OsResolverNameservers groups LAN entries before public entries. Preserve a usable
+// first target, including public DNS. Otherwise scan only the LAN group. Stop at
+// the public boundary, not at a particular IP: the public group can be synthetic.
+func pfProbeTarget(servers []string) string {
+	for i, server := range servers {
+		host, _, err := net.SplitHostPort(server)
+		if err != nil {
+			return ""
+		}
+		ip, err := netip.ParseAddr(host)
+		if err != nil {
+			return ""
+		}
+		lan := ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || tsaddr.CGNATRange().Contains(ip)
+		usable := ip.Is4() && !ip.IsLoopback() && !ip.IsUnspecified() && !ip.IsMulticast()
+		if usable && (i == 0 || lan) {
+			return host
+		}
+		if !lan {
+			return ""
+		}
+	}
+	return ""
+}
+
+// Keep the helper protocol bounded and free of resolver IDs, queries and raw OS errors.
+func pfProbeErrorCode(err error) string {
+	switch {
+	case errors.Is(err, syscall.ENETUNREACH), errors.Is(err, syscall.EHOSTUNREACH):
+		return "unreachable"
+	case errors.Is(err, syscall.EADDRNOTAVAIL):
+		return "source_unavailable"
+	case errors.Is(err, syscall.EACCES), errors.Is(err, syscall.EPERM):
+		return "permission"
+	default:
+		var ne net.Error
+		if errors.As(err, &ne) && ne.Timeout() {
+			return "timeout"
+		}
+		return "io"
+	}
+}
+
+func sendPFProbe(host, hexPacket string, dial func(string, string, time.Duration) (net.Conn, error), status io.Writer) error {
+	fail := func(stage string, err error) error {
+		_, _ = fmt.Fprintf(status, "%s:%s\n", stage, pfProbeErrorCode(err))
+		return err
+	}
+	packet, err := hex.DecodeString(hexPacket)
+	if err != nil {
+		_, _ = io.WriteString(status, "decode:invalid_argument\n")
+		return err
+	}
+	conn, err := dial("udp4", net.JoinHostPort(host, "53"), time.Second)
+	if err != nil {
+		return fail("dial", err)
+	}
+	defer conn.Close()
+	if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil {
+		return fail("deadline", err)
+	}
+	n, err := conn.Write(packet)
+	if err == nil && n != len(packet) {
+		err = io.ErrShortWrite
+	}
+	if err != nil {
+		return fail("write", err)
+	}
+	if _, err := io.WriteString(status, "sent\n"); err != nil {
+		return err
+	}
+	// Keep the socket alive for the local handler's response. A read failure does
+	// not undo the successful send acknowledged above.
+	_, _ = conn.Read(make([]byte, 512))
+	return nil
+}
+
+func parsePFProbeStatus(status string) pfProbeObservation {
+	if status == "sent\n" {
+		return pfProbeObservation{stage: "sent"}
+	}
+	stage, code, ok := strings.Cut(strings.TrimSuffix(status, "\n"), ":")
+	if ok && (stage == "decode" || stage == "dial" || stage == "deadline" || stage == "write") {
+		switch code {
+		case "unreachable", "source_unavailable", "permission", "timeout", "io", "invalid_argument":
+			return pfProbeObservation{stage: stage, code: code}
+		}
+	}
+	return pfProbeObservation{stage: "helper", code: "invalid_status"}
+}
+
+func pfProbeContextCode(ctx context.Context) string {
+	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+		return "timeout"
+	}
+	return "canceled"
+}
+
+// Receipt by the local handler is conclusive. A missing receipt is a failed
+// interception test only after the child confirms a successful packet send.
+func runPFProbe(ctx context.Context, cmd *exec.Cmd, received <-chan struct{}, deliveryTimeout time.Duration) pfProbeObservation {
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return pfProbeObservation{stage: "start", code: pfProbeErrorCode(err)}
+	}
+	if err := cmd.Start(); err != nil {
+		_ = stdout.Close()
+		return pfProbeObservation{stage: "start", code: pfProbeErrorCode(err)}
+	}
+	statusCh := make(chan pfProbeObservation, 1)
+	done := make(chan struct{})
+	go func() {
+		line, _ := bufio.NewReader(io.LimitReader(stdout, 128)).ReadString('\n')
+		statusCh <- parsePFProbeStatus(line)
+		_ = cmd.Wait()
+		close(done)
+	}()
+	defer func() {
+		_ = cmd.Process.Kill()
+		<-done
+	}()
+
+	var sent pfProbeObservation
+	select {
+	case <-received:
+		return pfProbeObservation{result: pfProbeIntercepted, stage: "received"}
+	case <-ctx.Done():
+		return pfProbeObservation{stage: "helper", code: pfProbeContextCode(ctx)}
+	case sent = <-statusCh:
+	}
+	if sent.stage != "sent" {
+		select {
+		case <-received:
+			return pfProbeObservation{result: pfProbeIntercepted, stage: "received"}
+		default:
+			return sent
+		}
+	}
+
+	timer := time.NewTimer(deliveryTimeout)
+	defer timer.Stop()
+	select {
+	case <-received:
+		return pfProbeObservation{result: pfProbeIntercepted, stage: "received"}
+	case <-ctx.Done():
+		return pfProbeObservation{stage: "delivery", code: pfProbeContextCode(ctx)}
+	case <-timer.C:
+		// Prefer receipt when notification and deadline became ready together.
+		select {
+		case <-received:
+			return pfProbeObservation{result: pfProbeIntercepted, stage: "received"}
+		default:
+			return pfProbeObservation{result: pfProbeNotIntercepted, stage: "delivery", code: "timeout"}
+		}
+	}
+}

cmd/cli/pf_probe_darwin_test.go

@@ -0,0 +1,213 @@
+//go:build darwin
+
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"os/exec"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestPFProbeDefaultCommand(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cmd := newPFProbeCommand(ctx, "192.0.2.1", "0102")
+	if !reflect.DeepEqual(cmd.Args, []string{os.Args[0], "pf-probe-send", "192.0.2.1", "0102"}) {
+		t.Fatalf("args=%q", cmd.Args)
+	}
+	if cmd.SysProcAttr == nil || cmd.SysProcAttr.Credential == nil {
+		t.Fatal("missing credentials")
+	}
+	cred := cmd.SysProcAttr.Credential
+	if cred.Uid != 0 || cred.Gid != 0 || cred.NoSetGroups || len(cred.Groups) != 0 {
+		t.Fatalf("unexpected credentials: %+v", cred)
+	}
+	if cmd.Cancel == nil {
+		t.Fatal("command is not context-bound")
+	}
+	// Do not start this command: credential changes require native privileged QA.
+}
+
+func TestPFProbeMonitorIndeterminateContinues(t *testing.T) {
+	oldProbe, oldReload, oldDelays := probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays
+	t.Cleanup(func() {
+		probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays = oldProbe, oldReload, oldDelays
+	})
+	pfInterceptMonitorDelays = []time.Duration{0, 0, 0}
+	probes, reloads := 0, 0
+	probePFInterceptFn = func(*prog) pfProbeObservation {
+		probes++
+		if probes == 2 {
+			return pfProbeObservation{result: pfProbeIntercepted}
+		}
+		return pfProbeObservation{stage: "dial", code: "unreachable"}
+	}
+	forceReloadPFInterceptFn = func(*prog) bool { reloads++; return true }
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}})
+	p.pfInterceptMonitor()
+	if probes != 3 || reloads != 0 || p.pfMonitorRunning.Load() {
+		t.Fatalf("probes=%d reloads=%d owner=%v", probes, reloads, p.pfMonitorRunning.Load())
+	}
+}
+
+func TestPFProbeRepairCorrelation(t *testing.T) {
+	for _, caller := range []string{"post_stabilization", "post_wake", "watchdog", "monitor"} {
+		for _, performed := range []bool{false, true} {
+			t.Run(caller+"/"+map[bool]string{false: "not_run", true: "completed"}[performed], func(t *testing.T) {
+				oldProbe, oldReload, oldDelays := probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays
+				t.Cleanup(func() {
+					probePFInterceptFn, forceReloadPFInterceptFn, pfInterceptMonitorDelays = oldProbe, oldReload, oldDelays
+				})
+				logs := captureTransitionLogs(t)
+				stubWakeProbeSchedule(t, 1)
+				pfInterceptMonitorDelays = []time.Duration{0}
+				p := sourceTestProg(&prog{dnsInterceptState: &pfState{}, stopCh: make(chan struct{})})
+				probes, reloads := 0, 0
+				probePFInterceptFn = func(*prog) pfProbeObservation {
+					probes++
+					if probes == 1 {
+						return pfProbeObservation{result: pfProbeNotIntercepted, probeID: "cause-probe", recoveryGeneration: 7, stage: "delivery", code: "timeout", target: "192.0.2.1:53"}
+					}
+					return pfProbeObservation{result: pfProbeIndeterminate, probeID: "confirmation-probe", recoveryGeneration: 8, stage: "dial", code: "unreachable"}
+				}
+				forceReloadPFInterceptFn = func(*prog) bool { reloads++; p.recoveryGen.Store(99); return performed }
+				switch caller {
+				case "post_stabilization":
+					p.verifyInterceptAfterStabilization()
+				case "post_wake":
+					p.runWakeProbeAttempt(0, 0)
+				case "watchdog":
+					p.checkPFWatchdogProbe(pfAnchorCheckIntact)
+				case "monitor":
+					p.pfInterceptMonitor()
+				}
+				expected := "not_run"
+				if performed {
+					expected = "reload_completed"
+				}
+				found, confirmation := false, false
+				for _, line := range strings.Split(logs.String(), "\n") {
+					var event map[string]interface{}
+					if json.Unmarshal([]byte(line), &event) != nil || event["message"] != "PF repair" {
+						continue
+					}
+					if event["probe_id"] != "cause-probe" || event["recovery_generation"] != float64(7) || event["caller"] != caller {
+						t.Fatalf("lost cause: %v", event)
+					}
+					if event["outcome"] == expected {
+						found = true
+					}
+					if event["outcome"] == "indeterminate" && event["confirmation_probe_id"] == "confirmation-probe" {
+						confirmation = true
+					}
+				}
+				if reloads != 1 || !found {
+					t.Fatalf("reloads=%d outcome found=%v", reloads, found)
+				}
+				if performed && caller != "watchdog" && !confirmation {
+					t.Fatal("missing confirmation outcome")
+				}
+			})
+		}
+	}
+}
+
+func TestPFProbeMissingTargetLogTransitions(t *testing.T) {
+	logs := captureTransitionLogs(t)
+	old := pfProbeNameservers
+	t.Cleanup(func() { pfProbeNameservers = old; pfProbeLogs = pfProbeLogState{} })
+	pfProbeLogs = pfProbeLogState{}
+	first := "[fe80::1%en0]:53"
+	pfProbeNameservers = func() []string { return []string{first, "76.76.2.0:53"} }
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}})
+	p.probePFIntercept()
+	p.probePFIntercept()
+	first = "127.0.0.1:53"
+	p.probePFIntercept()
+	var levels []string
+	for _, line := range strings.Split(logs.String(), "\n") {
+		var event map[string]interface{}
+		if json.Unmarshal([]byte(line), &event) != nil || event["message"] != "DNS intercept probe result" {
+			continue
+		}
+		levels = append(levels, event["level"].(string))
+		if len(levels) == 3 && event["repeated_results"] != float64(1) {
+			t.Fatal("missing suppressed count")
+		}
+	}
+	if !reflect.DeepEqual(levels, []string{"warn", "debug", "warn"}) {
+		t.Fatalf("levels=%v", levels)
+	}
+}
+
+func TestPFProbeUnsentDoesNotReload(t *testing.T) {
+	logs := captureTransitionLogs(t)
+	oldServers, oldCommand, oldReload := pfProbeNameservers, newPFProbeCommand, forceReloadPFInterceptFn
+	t.Cleanup(func() {
+		pfProbeNameservers, newPFProbeCommand, forceReloadPFInterceptFn = oldServers, oldCommand, oldReload
+	})
+	pfProbeNameservers = func() []string { return []string{"192.0.2.1:53"} }
+	newPFProbeCommand = func(ctx context.Context, host, packet string) *exec.Cmd { return pfProbeTestCommand(ctx, "dial") }
+	reloads := 0
+	forceReloadPFInterceptFn = func(*prog) bool { reloads++; return true }
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}, stopCh: make(chan struct{})})
+	if got := p.checkPFWatchdogProbe(pfAnchorCheckIntact); got != pfAnchorCheckDeferred {
+		t.Fatalf("watchdog result=%v", got)
+	}
+	p.verifyInterceptAfterStabilization()
+	stubWakeProbeSchedule(t, 1)
+	if restored, repaired := p.runWakeProbeAttempt(0, 0); restored || repaired {
+		t.Fatalf("unsent wake probe: restored=%v repaired=%v", restored, repaired)
+	}
+	if reloads != 0 {
+		t.Fatalf("unsent probes caused %d reloads", reloads)
+	}
+	output := logs.String()
+	for _, field := range []string{`"probe_id":`, `"recovery_generation":`, `"resolver_target":"192.0.2.1:53"`, `"stage":"dial"`, `"error_code":"unreachable"`, `"outcome":"indeterminate"`, `"repair_eligible":false`} {
+		if !strings.Contains(output, field) {
+			t.Errorf("missing %s in logs", field)
+		}
+	}
+	if strings.Contains(output, "interception is translating") || strings.Contains(output, "anchor was missing") {
+		t.Fatal("unknown probe claimed PF health or missing rules")
+	}
+}
+
+func TestPFProbeWatchdogMeaningfulFailureRepairsWithoutMissingAnchorClaim(t *testing.T) {
+	_, reloads := stubStabilizationProbe(t, []bool{false}, true)
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}})
+	if got := p.checkPFWatchdogProbe(pfAnchorCheckIntact); got != pfAnchorCheckDeferred {
+		t.Fatalf("probe repair must not be counted as missing-anchor restoration: %v", got)
+	}
+	if *reloads != 1 {
+		t.Fatalf("reloads=%d", *reloads)
+	}
+}
+
+func TestPFProbeWatchdogSkipsIndeterminateAnchor(t *testing.T) {
+	probes, reloads := stubStabilizationProbe(t, nil, true)
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}})
+	for _, r := range []pfAnchorCheckResult{pfAnchorCheckDeferred, pfAnchorCheckSkipped, pfAnchorCheckFailed, pfAnchorCheckRestored} {
+		if got := p.checkPFWatchdogProbe(r); got != r {
+			t.Fatalf("changed outcome %v -> %v", r, got)
+		}
+	}
+	if *probes != 0 || *reloads != 0 {
+		t.Fatal("probed without intact rule evidence")
+	}
+}
+
+func TestPFProbeNoIPv4TargetIsIndeterminate(t *testing.T) {
+	old := pfProbeNameservers
+	t.Cleanup(func() { pfProbeNameservers = old })
+	pfProbeNameservers = func() []string { return []string{"[fe80::1%en0]:53", "76.76.2.0:53"} }
+	p := sourceTestProg(&prog{dnsInterceptState: &pfState{}})
+	if got := p.probePFIntercept(); got.result != pfProbeIndeterminate {
+		t.Fatalf("got %v", got)
+	}
+}

cmd/cli/pf_probe_test.go

@@ -0,0 +1,292 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestPFProbeProcess(t *testing.T) {
+	mode := os.Getenv("CTRLD_PF_PROBE_TEST")
+	if mode == "" {
+		return
+	}
+	switch mode {
+	case "main_dispatch":
+		os.Args = []string{os.Args[0], "pf-probe-send", "127.0.0.1", "zz"}
+		Main()
+		os.Exit(99) // Production dispatch must exit on invalid input.
+	case "sent":
+		_, _ = os.Stdout.WriteString("sent\n")
+	case "dial":
+		_, _ = os.Stdout.WriteString("dial:unreachable\n")
+		os.Exit(1)
+	case "write":
+		_, _ = os.Stdout.WriteString("write:permission\n")
+		os.Exit(1)
+	case "malformed":
+		_, _ = os.Stdout.WriteString(strings.Repeat("private-payload", 1000))
+	case "hang":
+		time.Sleep(time.Minute)
+	case "silent":
+		os.Exit(1)
+	}
+	os.Exit(0)
+}
+
+func pfProbeTestCommand(ctx context.Context, mode string) *exec.Cmd {
+	cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestPFProbeProcess$")
+	cmd.Env = append(os.Environ(), "CTRLD_PF_PROBE_TEST="+mode)
+	return cmd
+}
+
+func TestPFProbeProcessClassification(t *testing.T) {
+	for _, tc := range []struct {
+		mode  string
+		want  pfProbeResult
+		stage string
+	}{
+		{"dial", pfProbeIndeterminate, "dial"},
+		{"write", pfProbeIndeterminate, "write"},
+		{"silent", pfProbeIndeterminate, "helper"},
+		{"malformed", pfProbeIndeterminate, "helper"},
+		{"sent", pfProbeNotIntercepted, "delivery"},
+	} {
+		t.Run(tc.mode, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			got := runPFProbe(ctx, pfProbeTestCommand(ctx, tc.mode), make(chan struct{}), 10*time.Millisecond)
+			if got.result != tc.want || got.stage != tc.stage {
+				t.Fatalf("got %+v, want %v/%s", got, tc.want, tc.stage)
+			}
+			if strings.Contains(got.code, "private") {
+				t.Fatal("helper output escaped into diagnostic")
+			}
+		})
+	}
+}
+
+func TestPFProbeStartFailureAndCancellation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	got := runPFProbe(ctx, exec.Command(t.TempDir()+"/missing-helper"), make(chan struct{}), time.Second)
+	if got.result != pfProbeIndeterminate || got.stage != "start" {
+		t.Fatalf("start failure: %+v", got)
+	}
+	ctx, cancel = context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	got = runPFProbe(ctx, pfProbeTestCommand(ctx, "hang"), make(chan struct{}), time.Second)
+	if got.result != pfProbeIndeterminate {
+		t.Fatalf("canceled helper must not request repair: %+v", got)
+	}
+}
+
+func TestPFProbeReceiptWins(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	received := make(chan struct{})
+	close(received)
+	got := runPFProbe(ctx, pfProbeTestCommand(ctx, "sent"), received, time.Millisecond)
+	if got.result != pfProbeIntercepted {
+		t.Fatalf("local receipt is conclusive: %+v", got)
+	}
+}
+
+type pfProbeErrorConn struct {
+	net.Conn
+	writeErr error
+	short    bool
+}
+
+func (c pfProbeErrorConn) SetDeadline(time.Time) error { return nil }
+func (c pfProbeErrorConn) Close() error                { return nil }
+func (c pfProbeErrorConn) Write(b []byte) (int, error) {
+	if c.short {
+		return len(b) - 1, nil
+	}
+	return 0, c.writeErr
+}
+
+func TestPFProbeSendFailures(t *testing.T) {
+	for _, tc := range []struct {
+		name, packet, want string
+		dial               func(string, string, time.Duration) (net.Conn, error)
+	}{
+		{"decode", "zz", "decode:invalid_argument\n", nil},
+		{"dial", "0102", "dial:unreachable\n", func(string, string, time.Duration) (net.Conn, error) {
+			return nil, &net.OpError{Op: "dial", Err: syscall.ENETUNREACH}
+		}},
+		{"write", "0102", "write:permission\n", func(string, string, time.Duration) (net.Conn, error) {
+			return pfProbeErrorConn{writeErr: syscall.EACCES}, nil
+		}},
+		{"short", "0102", "write:io\n", func(string, string, time.Duration) (net.Conn, error) { return pfProbeErrorConn{short: true}, nil }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var out bytes.Buffer
+			err := sendPFProbe("192.0.2.1", tc.packet, tc.dial, &out)
+			if err == nil || out.String() != tc.want {
+				t.Fatalf("status=%q err=%v", out.String(), err)
+			}
+			if strings.Contains(out.String(), "sent") {
+				t.Fatal("failed write claimed success")
+			}
+		})
+	}
+}
+
+func TestPFProbeSendRealUDP(t *testing.T) {
+	listener, err := net.ListenPacket("udp4", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+	received := make(chan []byte, 1)
+	go func() {
+		b := make([]byte, 100)
+		n, addr, e := listener.ReadFrom(b)
+		if e != nil {
+			return
+		}
+		received <- append([]byte(nil), b[:n]...)
+		_, _ = listener.WriteTo([]byte{0}, addr)
+	}()
+	var out bytes.Buffer
+	err = sendPFProbe("127.0.0.1", "010203", func(network, address string, timeout time.Duration) (net.Conn, error) {
+		if network != "udp4" || address != "127.0.0.1:53" {
+			t.Errorf("unexpected target %s %s", network, address)
+		}
+		return net.DialTimeout(network, listener.LocalAddr().String(), timeout)
+	}, &out)
+	if err != nil || out.String() != "sent\n" {
+		t.Fatalf("status=%q error=%v", out.String(), err)
+	}
+	select {
+	case b := <-received:
+		if !bytes.Equal(b, []byte{1, 2, 3}) {
+			t.Fatalf("wrong packet %v", b)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("helper reported sent without UDP receipt")
+	}
+}
+
+func TestPFProbeTarget(t *testing.T) {
+	for _, first := range []string{"[fe80::1%en0]:53", "[2001:db8::1]:53", "127.0.0.53:53", "0.0.0.0:53", "224.0.0.1:53", "bad"} {
+		// Resolver construction can append this public fallback. No later entry
+		// may become a new probe recipient when the original first target fails.
+		if got := pfProbeTarget([]string{first, "76.76.2.0:53"}); got != "" {
+			t.Fatalf("first=%s selected new recipient=%s", first, got)
+		}
+	}
+	for _, first := range []string{"192.0.2.10", "76.76.2.0"} {
+		if got := pfProbeTarget([]string{first + ":53", "192.0.2.11:53"}); got != first {
+			t.Fatalf("changed original first target %s to %s", first, got)
+		}
+	}
+	if got := pfProbeTarget(nil); got != "" {
+		t.Fatalf("empty list selected %s", got)
+	}
+}
+
+func TestPFProbeMainDispatch(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	out, err := pfProbeTestCommand(ctx, "main_dispatch").Output()
+	var exit *exec.ExitError
+	if !errors.As(err, &exit) || exit.ExitCode() != 1 || string(out) != "decode:invalid_argument\n" {
+		t.Fatalf("production Main dispatch: output=%q error=%v", out, err)
+	}
+}
+
+func TestPFProbeTargetLANScan(t *testing.T) {
+	for _, tc := range []struct {
+		servers []string
+		want    string
+	}{
+		{[]string{"[fe80::1%en0]:53", "192.168.1.1:53", "76.76.2.0:53"}, "192.168.1.1"},
+		{[]string{"127.0.0.1:53", "100.64.0.1:53", "76.76.2.0:53"}, "100.64.0.1"},
+		{[]string{"[fd00::1]:53", "169.254.1.1:53", "76.76.2.0:53"}, "169.254.1.1"},
+		{[]string{"[fe80::1%en0]:53", "76.76.2.0:53"}, ""},
+		{[]string{"[fe80::1%en0]:53", "8.8.8.8:53", "192.168.1.1:53"}, ""},
+		{[]string{"[2001:db8::1]:53", "192.168.1.1:53"}, ""},
+		{[]string{"76.76.2.0:53"}, "76.76.2.0"},
+		{[]string{"8.8.8.8:53", "192.168.1.1:53"}, "8.8.8.8"},
+	} {
+		if got := pfProbeTarget(tc.servers); got != tc.want {
+			t.Errorf("%v: got %q want %q", tc.servers, got, tc.want)
+		}
+	}
+}
+
+func TestPFProbeLogConcurrentRepeats(t *testing.T) {
+	var state pfProbeLogState
+	observed := pfProbeObservation{stage: "dial", code: "unreachable"}
+	state.change(observed)
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Add(1)
+		go func() { defer wg.Done(); state.change(observed) }()
+	}
+	wg.Wait()
+	observed.result, observed.stage, observed.code = pfProbeIntercepted, "received", ""
+	if retained, repeats := state.change(observed); !retained || repeats != 20 {
+		t.Fatalf("retained=%v repeats=%d", retained, repeats)
+	}
+}
+
+func TestPFProbeLogConditionChanges(t *testing.T) {
+	var state pfProbeLogState
+	missing := pfProbeObservation{stage: "target", code: "unavailable", target: "[fe80::1%en0]:53"}
+	if retain, repeats := state.change(missing); !retain || repeats != 0 {
+		t.Fatal("first missing target not retained")
+	}
+	missing.probeID, missing.recoveryGeneration = "new-attempt", 2
+	if retain, repeats := state.change(missing); retain || repeats != 1 {
+		t.Fatal("attempt IDs defeat dedupe")
+	}
+	for _, change := range []func(*pfProbeObservation){
+		func(o *pfProbeObservation) { o.target = "127.0.0.1:53" },
+		func(o *pfProbeObservation) { o.stage = "dial" },
+		func(o *pfProbeObservation) { o.code = "unreachable" },
+		func(o *pfProbeObservation) { o.result = pfProbeNotIntercepted },
+	} {
+		change(&missing)
+		if retain, repeats := state.change(missing); !retain || repeats != 1 {
+			t.Fatalf("change not retained with prior repeats: %v %d", retain, repeats)
+		}
+		if retain, repeats := state.change(missing); retain || repeats != 1 {
+			t.Fatal("repeat not coalesced")
+		}
+	}
+	healthy := pfProbeObservation{result: pfProbeIntercepted, stage: "received", target: missing.target}
+	if retain, repeats := state.change(healthy); !retain || repeats != 1 {
+		t.Fatal("restoration not retained")
+	}
+	if retain, _ := state.change(healthy); retain {
+		t.Fatal("normal receipt retained")
+	}
+}
+
+func TestPFProbeStatusIsBounded(t *testing.T) {
+	if got := parsePFProbeStatus("decode:invalid_argument\n"); got.stage != "decode" || got.code != "invalid_argument" || got.result != pfProbeIndeterminate {
+		t.Fatalf("invalid argument classification: %+v", got)
+	}
+	for _, s := range []string{"dial:secret-token\n", "secret:io\n", strings.Repeat("private", 1000)} {
+		got := parsePFProbeStatus(s)
+		if got.result != pfProbeIndeterminate || got.stage != "helper" || got.code != "invalid_status" {
+			t.Fatalf("untrusted status leaked: %+v", got)
+		}
+	}
+	if pfProbeErrorCode(errors.New("secret")) != "io" || pfProbeErrorCode(io.ErrShortWrite) != "io" {
+		t.Fatal("unbounded diagnostic")
+	}
+}

cmd/cli/prog.go

@@ -163,6 +163,14 @@ type prog struct {
 	recoveryRunning  atomic.Bool
 	recoveryGen      atomic.Uint64
 
+	// All deltas get an ID. Only accepted deltas replace recovery ownership.
+	networkTransitionGen    atomic.Uint64
+	networkAcceptedGen      atomic.Uint64
+	networkSourceMu         sync.Mutex
+	networkSourceState      *netmon.State
+	networkSourceEpoch      *netmon.State
+	networkSourceReadFailed bool
+
 	// recoveryDebounceTimer coalesces rapid NetworkChange recovery triggers
 	// into a single handleRecovery call. Only handleRecovery is debounced —
 	// all other state updates (IP, pf anchor, VPN DNS) run immediately.