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"
@@ -77,7 +78,8 @@ const (
 	// 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, pf interception is working. If not (timeout), rdr is broken.
+	// 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"
 
@@ -1477,10 +1479,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
@@ -1551,17 +1562,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
 	}
@@ -1600,8 +1623,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
@@ -1653,7 +1676,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)
 }
 
@@ -1680,7 +1703,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(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 {
@@ -1692,16 +1720,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
 	}
@@ -2011,18 +2042,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.
@@ -2070,6 +2090,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 rebuilds the pf anchor rules to exclude VPN DNS interfaces
 // and server IPs from interception. VPN DNS handlers using Network Extensions
 // (e.g., Tailscale MagicDNS) need DNS traffic to flow without any pf interference.
@@ -2144,67 +2185,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).
+		Dur("elapsed_ms", time.Since(started)).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.
@@ -2235,6 +2300,8 @@ func buildDNSQueryPacket(domain string) []byte {
 	return append(header, question...)
 }
 
+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).
@@ -2259,7 +2326,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 {
@@ -2270,22 +2337,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

@@ -1655,7 +1655,7 @@ func FlushDNSCache() error {
 
 // monitorNetworkChanges starts monitoring for network interface changes
 func (p *prog) monitorNetworkChanges() 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)
 		mainLog.Load().Printf("netmon: "+format, args...)
 	})
@@ -1664,183 +1664,7 @@ func (p *prog) monitorNetworkChanges() error {
 	}
 
 	mon.RegisterChangeCallback(func(delta *netmon.ChangeDelta) {
-		isMajorChange := mon.IsMajorChangeFrom(delta.Old, delta.New)
-		p.handleDNS64NetworkChange(delta, isMajorChange)
-
-		// Get map of valid interfaces
-		validIfaces := validInterfacesMap()
-
-		mainLog.Load().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
-					mainLog.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
-					mainLog.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
-			mainLog.Load().Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface)
-		}
-
-		if !changed {
-			mainLog.Load().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 {
-			mainLog.Load().Debug().Msg("No active interfaces found, skipping reinitialization")
-			return
-		}
-
-		mainLog.Load().Debug().Msg("Link state changed, re-bootstrapping")
-		for _, uc := range p.cfg.Upstream {
-			uc.ReBootstrap()
-		}
-
-		// Get IPs from default route interface in new state
-		selfIP := 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 {
-			mainLog.Load().Debug().Msgf("defaultRouteIP returned a non-IPv4 address: %s, ignoring it", selfIP)
-			selfIP = ""
-		}
-		var ipv6 string
-
-		if delta.New.DefaultRouteInterface != "" {
-			mainLog.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] {
-				ipAddr, _ := netip.ParsePrefix(ip.String())
-				addr := ipAddr.Addr()
-				if selfIP == "" && addr.Is4() {
-					mainLog.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
-			mainLog.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() {
-					mainLog.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()
-				}
-			}
-		}
-
-		// Only set the IPv4 default if selfIP is a valid IPv4 address.
-		if ip := net.ParseIP(selfIP); ip != nil && ip.To4() != nil {
-			ctrld.SetDefaultLocalIPv4(ip)
-			if !isMobile() && p.ciTable != nil {
-				p.ciTable.SetSelfIP(selfIP)
-			}
-		}
-		if ip := net.ParseIP(ipv6); ip != nil {
-			ctrld.SetDefaultLocalIPv6(ip)
-		}
-		mainLog.Load().Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6)
-
-		// we only trigger recovery flow for network changes on non router devices
-		if router.Name() == "" {
-			p.debounceRecovery()
-		}
-
-		// 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(true)
-			}
-			// Schedule delayed re-checks to catch async VPN teardown changes.
-			p.scheduleDelayedRechecks()
-		}
+		p.handleNetworkChange(delta, mon.IsMajorChangeFrom(delta.Old, delta.New))
 	})
 
 	mon.Start()
@@ -1848,6 +1672,269 @@ func (p *prog) monitorNetworkChanges() error {
 	return nil
 }
 
+// handleNetworkChange is the network monitor callback, kept separate for synthetic deltas.
+func (p *prog) handleNetworkChange(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()
+		mainLog.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 {
+				mainLog.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 {
+				mainLog.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(sourceState, transitionID)
+	after4, after6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	p.networkSourceMu.Unlock()
+
+	p.handleDNS64NetworkChange(delta, isMajorChange)
+	validIfaces := networkChangeValidInterfacesFn()
+
+	changed := false
+	changedInterface := ""
+	outcome := "ignored"
+	defer func() {
+		mainLog.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
+				mainLog.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
+				mainLog.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
+		mainLog.Load().Debug().Msgf("Default route changed from %s to %s", delta.Old.DefaultRouteInterface, delta.New.DefaultRouteInterface)
+	}
+
+	if !changed {
+		mainLog.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"
+		mainLog.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()
+
+	mainLog.Load().Debug().Msg("Link state changed, re-bootstrapping")
+	for _, uc := range p.cfg.Upstream {
+		uc.ReBootstrap()
+	}
+
+	// Get IPs from default route interface in new state
+	selfIP := networkChangeDefaultRouteIPFn()
+
+	// 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 {
+		mainLog.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 != "" {
+		mainLog.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() {
+				mainLog.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
+		mainLog.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() {
+				mainLog.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(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(ip)
+		if !isMobile() && p.ciTable != nil {
+			p.ciTable.SetSelfIP(selfIP)
+		}
+	}
+	if ip := net.ParseIP(ipv6); ip != nil && latestState != nil && sourceInvalidReason(latestState, ip) == "" {
+		ctrld.SetDefaultLocalIPv6(ip)
+	}
+	after4, after6 = ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	p.networkSourceMu.Unlock()
+	mainLog.Load().Debug().Msgf("Set default local IPv4: %s, IPv6: %s", selfIP, ipv6)
+
+	outcome = "accepted"
+	networkChangeReconcileFn(p, transitionID)
+}
+
+// reconcileNetworkChange is the first DNS/PF mutation boundary after source updates.
+func (p *prog) reconcileNetworkChange(transitionID uint64) {
+	// we only trigger recovery flow for network changes on non router devices
+	if router.Name() == "" {
+		p.debounceRecovery(transitionID)
+	}
+
+	// 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(true)
+		}
+		// 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.
@@ -2035,7 +2122,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()
 
@@ -2044,10 +2136,16 @@ func (p *prog) debounceRecovery() {
 		mainLog.Load().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)
 	})
 	mainLog.Load().Debug().Msg("Recovery debounce: scheduled (500ms window)")
 }
@@ -2057,13 +2155,28 @@ func (p *prog) debounceRecovery() {
 // upstream failure recoveries, waiting for recovery to complete (using a cancellable context without timeout),
 // and then re-applying the DNS settings.
 func (p *prog) handleRecovery(reason RecoveryReason) {
-	mainLog.Load().Debug().Msg("Starting recovery process: removing DNS settings")
+	p.handleRecoveryForTransition(reason, 0)
+}
 
-	recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason)
-	if !ok {
-		mainLog.Load().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()
+		mainLog.Load().Debug().Uint64("transition_id", transitionID).Msg("Recovery skipped: transition superseded")
 		return
 	}
+	recoveryCtx, gen, interceptRecovery, ok := p.beginRecovery(reason)
+	p.networkSourceMu.Unlock()
+	if !ok {
+		mainLog.Load().Debug().Uint64("transition_id", transitionID).
+			Uint64("recovery_generation", p.recoveryGen.Load()).
+			Msg("Upstream recovery already in progress; skipping duplicate trigger")
+		return
+	}
+	diagnostic := recoveryDiagnostic{transitionID: transitionID, generation: gen, reason: reason, started: time.Now()}
+	diagnostic.event(mainLog.Load().Debug()).Bool("intercept", interceptRecovery).Msg("Recovery begin")
+	outcome := "superseded"
+	defer func() { diagnostic.end(outcome) }()
 	if reason == RecoveryReasonNetworkChange {
 		mainLog.Load().Debug().Msg("Network change recovery now owns shared recovery state")
 	}
@@ -2125,7 +2238,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 		}
 	} else {
 		// Traditional flow: remove DNS settings to expose DHCP nameservers
-		p.resetDNS(false, false)
+		recoveryResetDNSFn(p, false, false)
 
 		// For an OS failure, reinitialize OS resolver nameservers immediately.
 		if reason == RecoveryReasonOSFailure {
@@ -2143,9 +2256,11 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 	upstreams := p.buildRecoveryUpstreams(reason)
 
 	// Wait indefinitely until one of the upstreams recovers.
-	recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams)
+	recovered, err := p.waitForUpstreamRecovery(recoveryCtx, upstreams, &diagnostic)
 	if err != nil {
-		mainLog.Load().Error().Err(err).Msg("Recovery canceled; DNS settings remain removed")
+		if p.recoveryOwnsState(gen) {
+			outcome = "canceled"
+		}
 		p.recoveryCanceledCleanup(gen)
 		return
 	}
@@ -2200,6 +2315,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 		mainLog.Load().Debug().Msgf("Recovery generation %d was superseded during completion; preserving successor state", gen)
 		return
 	}
+	outcome = "completed"
 	if interceptRecovery {
 		mainLog.Load().Info().Msg("DNS intercept recovery complete: disabling DHCP bypass, resuming normal flow")
 	}
@@ -2207,7 +2323,7 @@ func (p *prog) handleRecovery(reason RecoveryReason) {
 
 // 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()
 
@@ -2243,6 +2359,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
@@ -2330,27 +2447,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(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(nil)
-	}
+	validateDefaultLocalIPsFromDelta(newState, 0)
 }

cmd/cli/main.go

@@ -1,7 +1,6 @@
 package cli
 
 import (
-	"encoding/hex"
 	"io"
 	"net"
 	"os"
@@ -214,18 +213,7 @@ func initCache() {
 //
 // Usage: ctrld pf-probe-send <host> <hex-encoded-dns-packet>
 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)
-	// Read response (don't care about result, just need the send to happen)
-	buf := make([]byte, 512)
-	_, _ = conn.Read(buf)
 }

cmd/cli/network_monitor_test.go

@@ -0,0 +1,52 @@
+package cli
+
+import (
+	"net"
+	"testing"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+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(net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(net.ParseIP("2001:db8::10"))
+	calls := 0
+	networkChangeReconcileFn = func(*prog, uint64) {
+		calls++
+		if ctrld.GetDefaultLocalIPv4() != nil {
+			t.Fatal("departed IPv4 survived to reconciliation")
+		}
+	}
+	p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}
+	if err := p.monitorNetworkChanges(); 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,32 @@
+package cli
+
+import (
+	"errors"
+	"net"
+	"testing"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+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(net.ParseIP("192.0.2.1"))
+	ctrld.SetDefaultLocalIPv6(nil)
+	p := &prog{cfg: &ctrld.Config{}, networkSourceEpoch: major}
+	calls := 0
+	networkChangeReconcileFn = func(*prog, uint64) { calls++ }
+	p.handleNetworkChange(&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,200 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/rs/zerolog"
+	"github.com/Control-D-Inc/ctrld"
+	ctrldnet "github.com/Control-D-Inc/ctrld/internal/net"
+	"tailscale.com/net/netmon"
+)
+
+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  = validInterfacesMap
+	networkChangeDefaultRouteIPFn   = 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(state *netmon.State, transitionID uint64) {
+	before4, before6 := ctrld.GetDefaultLocalIPv4(), ctrld.GetDefaultLocalIPv6()
+	reason4, reason6 := sourceInvalidReason(state, before4), sourceInvalidReason(state, before6)
+	if reason4 != "" {
+		ctrld.SetDefaultLocalIPv4(nil)
+	}
+	if reason6 != "" {
+		ctrld.SetDefaultLocalIPv6(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.
+	mainLog.Load().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 {
+	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 *zerolog.Event) *zerolog.Event {
+	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(mainLog.Load().Warn()).Str("failure", failure).
+			Msg("Recovery waiting for upstream; first failure")
+	})
+}
+
+func (d *recoveryDiagnostic) end(outcome string) {
+	e := mainLog.Load().Debug()
+	if outcome == "canceled" || d.failed.Load() {
+		e = mainLog.Load().Warn()
+	}
+	d.event(e).Str("outcome", outcome).Bool("had_failure", d.failed.Load()).Dur("duration_ms", time.Since(d.started)).Msg("Recovery end")
+}

cmd/cli/network_transition_concurrency_test.go

@@ -0,0 +1,87 @@
+package cli
+
+import (
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+func TestNetworkChangeOverlappingCallbackCannotRestoreRemovedSource(t *testing.T) {
+	sourceTestGlobals(t)
+	logs := captureTransitionLogs(t)
+	ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(nil)
+	entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	networkChangeDefaultRouteIPFn = func() string {
+		close(entered)
+		<-release
+		return "192.0.2.10"
+	}
+	reconciliations := 0
+	networkChangeReconcileFn = func(*prog, uint64) { reconciliations++ }
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) {}
+	p := &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(&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(&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(nil)
+	ctrld.SetDefaultLocalIPv6(nil)
+	networkChangeReconcileFn = func(*prog, uint64) { t.Fatal("old snapshot reconciled") }
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { t.Fatal("old snapshot reconciled") }
+	p := &prog{cfg: &ctrld.Config{}}
+	p.handleNetworkChange(&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 := &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,94 @@
+package cli
+
+import (
+	"errors"
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+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(net.ParseIP(initial))
+			ctrld.SetDefaultLocalIPv6(nil)
+			p := &prog{cfg: &ctrld.Config{}}
+			reconciled := 0
+			networkChangeReconcileFn = func(*prog, 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(&netmon.ChangeDelta{Monitor: mon, Old: major, New: latest}, false)
+			olderMinor := *major
+			p.handleNetworkChange(&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(&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() map[string]struct{} { return map[string]struct{}{} }
+	calls := 0
+	networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ }
+	p := &prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}
+	p.handleNetworkChange(&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,128 @@
+package cli
+
+import (
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	ctrld "github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+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() map[string]struct{} { return map[string]struct{}{} }
+			calls := 0
+			networkChangeIgnoredInterceptFn = func(*prog, *netmon.ChangeDelta, time.Time) { calls++ }
+			p := &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(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 := &prog{cfg: &ctrld.Config{}}
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	p.handleNetworkChange(&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(&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(nil)
+	ctrld.SetDefaultLocalIPv6(nil)
+	entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	networkChangeDefaultRouteIPFn = func() string { close(entered); <-release; return "192.0.2.20" }
+	reconciled := make(chan uint64, 2)
+	networkChangeReconcileFn = func(_ *prog, id uint64) { reconciled <- id }
+	p := &prog{cfg: &ctrld.Config{}}
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	go func() {
+		defer close(done)
+		p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true)
+	}()
+	<-entered
+	p.handleNetworkChange(&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() string {
+		if lookups.Add(1) == 1 {
+			close(entered)
+			<-release
+			return "192.0.2.20"
+		}
+		return ""
+	}
+	reconciled := make(chan uint64, 2)
+	networkChangeReconcileFn = func(_ *prog, id uint64) { reconciled <- id }
+	p := &prog{cfg: &ctrld.Config{}}
+	a := sourceTestState("en0", true, "192.0.2.20/24")
+	go func() {
+		defer close(done)
+		p.handleNetworkChange(&netmon.ChangeDelta{Old: sourceTestState("en0", true, "192.0.2.10/24"), New: a}, true)
+	}()
+	<-entered
+	p.handleNetworkChange(&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 := &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,308 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"net/netip"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+	"github.com/Control-D-Inc/ctrld"
+	"tailscale.com/net/netmon"
+)
+
+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 := zerolog.New(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() map[string]struct{} { return map[string]struct{}{"en0": {}} }
+	networkChangeDefaultRouteIPFn = func() string { return "" }
+	t.Cleanup(func() {
+		ctrld.SetDefaultLocalIPv4(v4)
+		ctrld.SetDefaultLocalIPv6(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(net.ParseIP(v4))
+			ctrld.SetDefaultLocalIPv6(net.ParseIP(v6))
+			networkChangeDefaultRouteIPFn = func() string { return tt.preferred }
+			p := &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, id uint64) {
+				accepted++
+				if id != 1 {
+					t.Errorf("transition ID = %d", id)
+				}
+				assertSources()
+			}
+			networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { ignored++; assertSources() }
+			p.handleNetworkChange(&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(net.ParseIP(v4))
+	ctrld.SetDefaultLocalIPv6(net.ParseIP(v6))
+	networkChangeDefaultRouteIPFn = func() 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, _ 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") }
+	(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}).handleNetworkChange(&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(net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(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, _ uint64) { t.Fatal("unexpected recovery") }
+	networkChangeIgnoredInterceptFn = func(_ *prog, _ *netmon.ChangeDelta, _ time.Time) { called++ }
+	(&prog{cfg: &ctrld.Config{}, dnsInterceptState: &interceptStateStub{}}).handleNetworkChange(&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 := &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 := &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))
+	logger := zerolog.New(zerolog.MultiLevelWriter(debug, &zerolog.FilteredLevelWriter{Writer: zerolog.LevelWriterAdapter{Writer: warnings}, Level: zerolog.WarnLevel}))
+	mainLog.Store(&logger)
+	t.Cleanup(func() { mainLog.Store(old) })
+	ctrld.SetDefaultLocalIPv4(net.ParseIP("192.0.2.10"))
+	ctrld.SetDefaultLocalIPv6(nil)
+	state := sourceTestState("en0", false, "192.0.2.10/24")
+	for range 100 {
+		validateDefaultLocalIPsFromDelta(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 := (&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())
+	}
+}

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 := &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
+				})
+				stubWakeProbeSchedule(t, 1)
+				pfInterceptMonitorDelays = []time.Duration{0}
+				p := &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 }
+				before := len(logOutput.String())
+				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(logOutput.String()[before:], "\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) {
+	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 := &prog{dnsInterceptState: &pfState{}}
+	before := len(logOutput.String())
+	p.probePFIntercept()
+	p.probePFIntercept()
+	first = "127.0.0.1:53"
+	p.probePFIntercept()
+	var levels []string
+	for _, line := range strings.Split(logOutput.String()[before:], "\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) {
+	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 := &prog{dnsInterceptState: &pfState{}, stopCh: make(chan struct{})}
+	before := len(logOutput.String())
+	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)
+	}
+	logs := logOutput.String()[before:]
+	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(logs, field) {
+			t.Errorf("missing %s in logs", field)
+		}
+	}
+	if strings.Contains(logs, "interception is translating") || strings.Contains(logs, "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 := &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 := &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 := &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

@@ -161,6 +161,14 @@ type prog struct {
 	// only touches shared recovery state if it is still the newest (#597).
 	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.

docs/dns-intercept-mode.md

@@ -312,12 +312,14 @@ Programs like Parallels Desktop reload `/etc/pf.conf` when creating/destroying v
 
 ctrld detects interface appearance/disappearance and spawns an async probe monitor:
 
-1. **Probe mechanism:** A subprocess runs with GID=0 (wheel, not `_ctrld`) and sends a DNS query to the OS resolver. If pf interception is working, the query gets redirected to ctrld (127.0.0.1:53) and is detected in the DNS handler. If broken, it times out after 1s.
+1. **Probe mechanism:** A subprocess runs with GID=0 (wheel, not `_ctrld`) and sends a DNS query to the OS resolver. If pf interception is working, the query gets redirected to ctrld (127.0.0.1:53) and is detected in the DNS handler. A confirmed send without local receipt within 1s gives `not_intercepted`.
 2. **Backoff schedule:** Probes at 0, 0.5, 1, 2, 4 seconds (~8s window) to win the race against async pf reloads by the hypervisor. Only one monitor runs at a time (atomic singleton).
-3. **Auto-heal:** On probe failure, `forceReloadPFMainRuleset()` dumps the running ruleset and pipes it back through `pfctl -f -`, resetting pf's translation engine. VPN-safe because it reassembles from the current running state.
+3. **Auto-heal:** Only after `not_intercepted`, `forceReloadPFMainRuleset()` dumps the running ruleset and pipes it back through `pfctl -f -`, resetting pf's translation engine. VPN-safe because it reassembles from the current running state.
 4. **Watchdog integration:** The 30s watchdog also runs the probe when rule text checks pass, as a safety net for unknown corruption causes.
 
-This approach detects **actual broken DNS** rather than guessing from trigger events, making it robust against future unknown corruption scenarios.
+The probe uses IPv4 only. Missing targets and helper failures give `indeterminate`, which does not cause a reload.
+See [target selection and diagnostics](network-recovery-diagnostics.md#pf-probes-macos).
+A successful probe proves local interception delivery, not remote DNS health.
 
 ### 5. Proactive DoH Connection Pool Reset
 

docs/network-recovery-diagnostics.md

@@ -0,0 +1,88 @@
+# Network-recovery diagnostics
+
+Use continuous `ctrld log tail` capture during a reproduction. Internal logs rotate by size.
+A later `log send` can contain recent debug events and older warnings.
+Match timestamps and IDs. The last retained error does not necessarily describe the current network.
+Issue #604 tracks upload-time device and version headers separately.
+
+## Transitions and recovery
+
+`Network transition` debug events include a `transition_id`, affected interface, default route, and stored source addresses before and after the callback.
+The outcome identifies an accepted, ignored, superseded, or unusable change.
+Newer accepted changes supersede pending recovery. Ignored changes do not cancel accepted recovery.
+Source commits validate addresses against the current network state.
+Netmon does not order minor callbacks within one major cache epoch.
+For minor callbacks and late major callbacks in that epoch, ctrld reads current interface addresses and flags.
+This read does not change host DNS. A failed read defers source writes, but reconciliation continues.
+
+`Removed stale resolver source` is a warning only when ctrld removes an invalid address.
+The warning gives the reason and survives debug rotation.
+`Recovery begin` and `Recovery end` connect the transition to a `recovery_generation`, reason, outcome, and elapsed time.
+Only the first upstream failure in each recovery produces a bounded summary.
+If a recovery encounters a failure, its final outcome also enters the warning stream.
+Normal successful recovery stays at debug level.
+
+These summaries exclude upstream URLs and queried names.
+Each PF probe records the recovery generation at its start.
+A later recovery cannot receive credit for an earlier probe.
+
+### Cancellation event migration
+
+`Recovery end` with `outcome=canceled` replaces `Recovery canceled; DNS settings remain removed`.
+Update support runbooks and log searches to use the structured event.
+The cancellation event alone does not prove that interface DNS remains removed.
+
+## PF probes (macOS)
+
+`DNS intercept probe result` includes these fields:
+
+- `probe_id` and `recovery_generation`
+- Numeric `resolver_target` and `resolver_family`
+- `stage`, bounded `error_code`, and `outcome`
+- `repair_eligible`, `repeated_results`, and elapsed milliseconds
+
+A changed unsuccessful condition produces a warning. The condition consists of the stage, error code, outcome, and target.
+Equal results stay at debug level. The next changed condition includes the preceding repeat count.
+The first restored receipt also produces a warning-stream event. This event prevents an older failure from appearing current.
+Normal receipts and shutdown cancellation stay at debug level.
+The warning buffer is finite. These events do not guarantee indefinite retention.
+
+The outcomes have these meanings:
+
+- `intercepted`: the registered query reached the local handler. This proves interception delivery, not remote DNS health.
+- `not_intercepted`: the helper confirmed a successful UDP write, but the handler did not receive the query before the deadline.
+  This result permits bounded repair. It does not identify an external PF writer or explain the delivery failure.
+- `indeterminate`: the probe has no suitable target, cannot send, receives an invalid helper acknowledgement, or stops before a conclusive result.
+  This result alone does not cause a PF reload or transport reset. The existing monitor or watchdog retries later.
+  Independent evidence of missing rules still permits ordinary repair.
+
+The literal stages are `target`, `start`, `decode`, `dial`, `deadline`, `write`, `sent`, `helper`, `delivery`, and `received`.
+`sent` is the helper acknowledgement before the final delivery result.
+The error codes are `unavailable`, `invalid_argument`, `unreachable`, `source_unavailable`, `permission`, `timeout`, `io`, `invalid_status`, and `canceled`.
+Successful results have an empty error code. No raw subprocess output enters these fields.
+
+The sender runs outside the `_ctrld` group exemption.
+ctrld blocks outgoing IPv6 DNS with its own PF rule and does not redirect it.
+The selector preserves a usable first IPv4 target, including an original public resolver.
+Otherwise, it selects the first usable IPv4 address in the leading LAN group.
+It stops at the public group, which can contain a synthetic fallback. It does not blacklist a particular public IP.
+Loopback, unspecified, and multicast addresses are not usable targets.
+
+`PF repair` events carry the causal `probe_id` and `recovery_generation` for all four callers.
+They record `started`, `not_run`, or `reload_completed`. Completed means that the reload ran, not that interception recovered.
+A confirming probe adds its own ID and its interception outcome to the original repair correlation.
+The watchdog leaves confirmation to its next scheduled probe.
+`budget_exhausted` records a wake repair that the existing budget prevents.
+A probe-triggered reload does not count as a missing-anchor repair.
+
+See [PF DNS interception](pf-dns-intercept.md) and [DNS intercept mode](dns-intercept-mode.md) for the rule lifecycle.
+
+## Reproduction boundaries
+
+Compare the affected release with the exact candidate artifact on an owned host.
+Keep the configuration unchanged. Capture ordinary DNS and a direct query to the actual listener and port.
+Measure recovery after a usable route and resolver return. Exclude unavoidable offline time.
+Record source-address changes, PF observations, and upstream results. Success at one layer does not prove success at another.
+
+Use the recovery mode, ownership, outcome, and effective network-service DNS configuration to assess cancellation.
+Never flush global PF state to collect diagnostics.

docs/pf-dns-intercept.md

@@ -278,10 +278,13 @@ change handler and spawns an asynchronous interception probe monitor:
 1. A subprocess sends a DNS query WITHOUT the `_ctrld` group GID, so pf
    intercept rules apply to it
 2. If ctrld receives the query → pf interception is working
-3. If the query times out (1s) → pf translation is broken
-4. On failure: `forceReloadPFMainRuleset()` does `pfctl -f -` with the current
+3. A successful UDP write without local receipt within 1s gives `not_intercepted`.
+4. Only this meaningful failure permits probe-triggered repair: `forceReloadPFMainRuleset()` does `pfctl -f -` with the current
    running ruleset, resetting pf's translation engine
 
+A missing target or helper failure gives `indeterminate` and does not cause a reload.
+The probe uses IPv4 only. See [target selection and diagnostics](network-recovery-diagnostics.md#pf-probes-macos).
+
 The monitor probes with exponential backoff (0, 0.5, 1, 2, 4s) to win the race
 against async pf reloads. Only one monitor runs at a time (singleton). The
 watchdog also runs the probe every 30s as a safety net.
@@ -300,8 +303,8 @@ Some pf.conf files include `set skip on lo0` which tells pf to skip ALL processi
 
 **Mitigation:** the interception probe. `probePFIntercept()` sends a real query from
 outside the `_ctrld` group and confirms the listener received the redirect, which cannot
-succeed while pf is bypassing loopback — so a skip on `lo0` shows up as a probe failure
-and triggers a full reload.
+succeed while pf bypasses loopback. A confirmed send without receipt permits a full reload.
+An unsent or unavailable probe is indeterminate and does not trigger repair.
 
 **Not implemented, contrary to earlier versions of this document:** ctrld does *not*
 strip `lo0` from `set skip on` directives, and the watchdog does *not* inspect skip